Repository: nats-io/nats-server
Pull request #3364 (2022-08-11 21:27:13)

fixed consumer source update receiving upstream msgs
if an origin stream contains:
1M msgs with subject foo and 1M msgs with subject bar

IF the source consumer changes their filter from foo to bar
Then it would have received messages for subject bar.
This happens because this tail was filtered and their
respective seqno was not communicated to the consumer.

This is somewhat unexpected. It is also coincidental.
Had the last message in the stream had subject foo then
this wouldn't happen.

Therefore, when completely changing the subject say,
from foo to bar, we only receive messages received
after the time the change was made.

However, if the old and new subject overlap in any way,
we go by sequence number. Meaning in these cases the
outlined behavior remains in order to not induce artificial
message loss for the part of the subject space that is
covered by old and new filter.

Signed-off-by: Matthias Hanel <mh@synadia.com>

 - [ ] Link to issue, e.g. `Resolves #NNN`
 - [ ] Documentation added (if applicable)
 - [ ] Tests added
 - [ ] Branch rebased on top of current main (`git pull --rebase origin main`)
 - [ ] Changes squashed to a single commit (described [here](http://gitready.com/advanced/2009/02/10/squashing-commits-with-rebase.html))
 - [ ] Build is green in Travis CI
 - [ ] You have certified that the contribution is your original work and that you license the work to the project under the [Apache 2 license](https://github.com/nats-io/nats-server/blob/main/LICENSE)

Resolves #

### Changes proposed in this pull request:

 -
 -
 -

/cc @nats-io/core



Patch:
diff --git a/dependencies.md b/dependencies.md
index 294ee8c2..bf328f5e 100644
--- a/dependencies.md
+++ b/dependencies.md
@@ -6,12 +6,15 @@ This file lists the dependencies used in this repository.
 |-|-|
 | Go | BSD 3-Clause "New" or "Revised" License |
 | github.com/nats-io/nats-server/v2 | Apache License 2.0 |
-| github.com/golang/protobuf v1.3.5 |  |
-| github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c | Apache License 2.0 |
-| github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0 | Apache License 2.0 |
-| github.com/nats-io/nkeys v0.2.0 | Apache License 2.0 |
+| github.com/golang/protobuf v1.4.2 | BSD 3-Clause "New" or "Revised" License |
+| github.com/nats-io/jwt/v2 v2.3.0 | Apache License 2.0 |
+| github.com/nats-io/nats.go v1.16.1-0.20220810192301-fb5ca2cbc995 | Apache License 2.0 |
+| github.com/nats-io/nkeys v0.3.0 | Apache License 2.0 |
 | github.com/nats-io/nuid v1.0.1 | Apache License 2.0 |
-| golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897 | BSD 3-Clause "New" or "Revised" License |
-| golang.org/x/sys v0.0.0-20191022100944-742c48ecaeb7 | BSD 3-Clause "New" or "Revised" License |
-| golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 | BSD 3-Clause "New" or "Revised" License |
-| github.com/minio/highwayhash v1.0.0 | Apache License 2.0 |
+| golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e | BSD 3-Clause "New" or "Revised" License |
+| golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a | BSD 3-Clause "New" or "Revised" License |
+| golang.org/x/time v0.0.0-20220411224347-583f2d630306 | BSD 3-Clause "New" or "Revised" License |
+| google.golang.org/protobuf v1.23.0 | BSD 3-Clause "New" or "Revised" License |
+| github.com/klauspost/compress/s2 v1.15.9 | BSD 3-Clause "New" or "Revised" License |
+| github.com/minio/highwayhash v1.0.2 | Apache License 2.0 |
+| go.uber.org/automaxprocs v1.5.1 | MIT License |
diff --git a/server/auth.go b/server/auth.go
index d2295c39..ffc9de94 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -371,6 +371,161 @@ func (c *client) matchesPinnedCert(tlsPinnedCerts PinnedCertSet) bool {
 	return true
 }
 
+func processUserPermissionsTemplate(lim jwt.UserPermissionLimits, ujwt *jwt.UserClaims, acc *Account) (jwt.UserPermissionLimits, error) {
+	nArrayCartesianProduct := func(a ...[]string) [][]string {
+		c := 1
+		for _, a := range a {
+			c *= len(a)
+		}
+		if c == 0 {
+			return nil
+		}
+		p := make([][]string, c)
+		b := make([]string, c*len(a))
+		n := make([]int, len(a))
+		s := 0
+		for i := range p {
+			e := s + len(a)
+			pi := b[s:e]
+			p[i] = pi
+			s = e
+			for j, n := range n {
+				pi[j] = a[j][n]
+			}
+			for j := len(n) - 1; j >= 0; j-- {
+				n[j]++
+				if n[j] < len(a[j]) {
+					break
+				}
+				n[j] = 0
+			}
+		}
+		return p
+	}
+	applyTemplate := func(list jwt.StringList, failOnBadSubject bool) (jwt.StringList, error) {
+		found := false
+	FOR_FIND:
+		for i := 0; i < len(list); i++ {
+			// check if templates are present
+			for _, tk := range strings.Split(list[i], tsep) {
+				if strings.HasPrefix(tk, "{{") && strings.HasSuffix(tk, "}}") {
+					found = true
+					break FOR_FIND
+				}
+			}
+		}
+		if !found {
+			return list, nil
+		}
+		// process the templates
+		emittedList := make([]string, 0, len(list))
+		for i := 0; i < len(list); i++ {
+			tokens := strings.Split(list[i], tsep)
+
+			newTokens := make([]string, len(tokens))
+			tagValues := [][]string{}
+
+			for tokenNum, tk := range tokens {
+				if strings.HasPrefix(tk, "{{") && strings.HasSuffix(tk, "}}") {
+					op := strings.ToLower(strings.TrimSuffix(strings.TrimPrefix(tk, "{{"), "}}"))
+					switch {
+					case op == "name()":
+						tk = ujwt.Name
+					case op == "subject()":
+						tk = ujwt.Subject
+					case op == "account-name()":
+						acc.mu.RLock()
+						name := acc.nameTag
+						acc.mu.RUnlock()
+						tk = name
+					case op == "account-subject()":
+						tk = ujwt.IssuerAccount
+					case (strings.HasPrefix(op, "tag(") || strings.HasPrefix(op, "account-tag(")) &&
+						strings.HasSuffix(op, ")"):
+						// insert dummy tav value that will throw of subject validation (in case nothing is found)
+						tk = _EMPTY_
+						// collect list of matching tag values
+
+						var tags jwt.TagList
+						var tagPrefix string
+						if strings.HasPrefix(op, "account-tag(") {
+							acc.mu.RLock()
+							tags = acc.tags
+							acc.mu.RUnlock()
+							tagPrefix = fmt.Sprintf("%s:", strings.ToLower(
+								strings.TrimSuffix(strings.TrimPrefix(op, "account-tag("), ")")))
+						} else {
+							tags = ujwt.Tags
+							tagPrefix = fmt.Sprintf("%s:", strings.ToLower(
+								strings.TrimSuffix(strings.TrimPrefix(op, "tag("), ")")))
+						}
+
+						valueList := []string{}
+						for _, tag := range tags {
+							if strings.HasPrefix(tag, tagPrefix) {
+								tagValue := strings.TrimPrefix(tag, tagPrefix)
+								valueList = append(valueList, tagValue)
+							}
+						}
+						if len(valueList) != 0 {
+							tagValues = append(tagValues, valueList)
+						}
+					default:
+						// if macro is not recognized, throw off subject check on purpose
+						tk = " "
+					}
+				}
+				newTokens[tokenNum] = tk
+			}
+			// fill in tag value placeholders
+			if len(tagValues) == 0 {
+				emitSubj := strings.Join(newTokens, tsep)
+				if IsValidSubject(emitSubj) {
+					emittedList = append(emittedList, emitSubj)
+				} else if failOnBadSubject {
+					return nil, fmt.Errorf("generated invalid subject")
+				}
+				// else skip emitting
+			} else {
+				// compute the cartesian product and compute subject to emit for each combination
+				for _, valueList := range nArrayCartesianProduct(tagValues...) {
+					b := strings.Builder{}
+					for i, token := range newTokens {
+						if token == _EMPTY_ {
+							b.WriteString(valueList[0])
+							valueList = valueList[1:]
+						} else {
+							b.WriteString(token)
+						}
+						if i != len(newTokens)-1 {
+							b.WriteString(tsep)
+						}
+					}
+					emitSubj := b.String()
+					if IsValidSubject(emitSubj) {
+						emittedList = append(emittedList, emitSubj)
+					} else if failOnBadSubject {
+						return nil, fmt.Errorf("generated invalid subject")
+					}
+					// else skip emitting
+				}
+			}
+		}
+		return emittedList, nil
+	}
+	var err error
+	if lim.Permissions.Sub.Allow, err = applyTemplate(lim.Permissions.Sub.Allow, false); err != nil {
+		return jwt.UserPermissionLimits{}, err
+	} else if lim.Permissions.Sub.Deny, err = applyTemplate(lim.Permissions.Sub.Deny, true); err != nil {
+		return jwt.UserPermissionLimits{}, err
+	} else if lim.Permissions.Pub.Allow, err = applyTemplate(lim.Permissions.Pub.Allow, false); err != nil {
+		return jwt.UserPermissionLimits{}, err
+	} else if lim.Permissions.Pub.Deny, err = applyTemplate(lim.Permissions.Pub.Deny, true); err != nil {
+		return jwt.UserPermissionLimits{}, err
+	}
+	return lim, nil
+}
+
 func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) bool {
 	var (
 		nkey *NkeyUser
@@ -621,8 +776,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			} else if uSc, ok := scope.(*jwt.UserScope); !ok {
 				c.Debugf("User JWT is not valid")
 				return false
-			} else {
-				juc.UserPermissionLimits = uSc.Template
+			} else if juc.UserPermissionLimits, err = processUserPermissionsTemplate(uSc.Template, juc, acc); err != nil {
+				c.Debugf("User JWT generated invalid permissions")
+				return false
 			}
 		}
 		if acc.IsExpired() {
diff --git a/server/const.go b/server/const.go
index 7cb3380c..67d36d86 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.9.0-RC.2"
+	VERSION = "2.9.0-RC.5"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index ccaeeef1..75b585fe 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1958,7 +1958,7 @@ func (o *consumer) checkRedelivered(slseq uint64) {
 	}
 	var shouldUpdateState bool
 	for sseq := range o.rdc {
-		if sseq < o.asflr || sseq > lseq {
+		if sseq < o.asflr || (lseq > 0 && sseq > lseq) {
 			delete(o.rdc, sseq)
 			o.removeFromRedeliverQueue(sseq)
 			shouldUpdateState = true
@@ -4031,6 +4031,10 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 		if dflag {
 			n.Delete()
 		} else {
+			// Try to install snapshot on clean exit
+			if snap, err := o.store.EncodedState(); err == nil {
+				n.InstallSnapshot(snap)
+			}
 			n.Stop()
 		}
 	}
diff --git a/server/errors.json b/server/errors.json
index 5730766e..fc5f9327 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1220,10 +1220,10 @@
     "deprecates": ""
   },
   {
-    "constant": "JSStreamMoveInProgress",
+    "constant": "JSStreamMoveInProgressF",
     "code": 400,
     "error_code": 10124,
-    "description": "stream move already in progress",
+    "description": "stream move already in progress: {msg}",
     "comment": "",
     "help": "",
     "url": "",
diff --git a/server/filestore.go b/server/filestore.go
index 21ee18f1..35574cf5 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -16,6 +16,7 @@ package server
 import (
 	"archive/tar"
 	"bytes"
+	"crypto/aes"
 	"crypto/cipher"
 	"crypto/rand"
 	"crypto/sha256"
@@ -55,6 +56,8 @@ type FileStoreConfig struct {
 	SyncInterval time.Duration
 	// AsyncFlush allows async flush to batch write operations.
 	AsyncFlush bool
+	// Cipher is the cipher to use when encrypting.
+	Cipher StoreCipher
 }
 
 // FileStreamInfo allows us to remember created time.
@@ -63,6 +66,24 @@ type FileStreamInfo struct {
 	StreamConfig
 }
 
+type StoreCipher int
+
+const (
+	ChaCha StoreCipher = iota
+	AES
+)
+
+func (cipher StoreCipher) String() string {
+	switch cipher {
+	case ChaCha:
+		return "ChaCha20-Poly1305"
+	case AES:
+		return "AES-GCM"
+	default:
+		return "Unknown StoreCipher"
+	}
+}
+
 // File ConsumerInfo is used for creating consumer stores.
 type FileConsumerInfo struct {
 	Created time.Time
@@ -113,7 +134,7 @@ type msgBlock struct {
 	mu      sync.RWMutex
 	fs      *fileStore
 	aek     cipher.AEAD
-	bek     *chacha20.Cipher
+	bek     cipher.Stream
 	seed    []byte
 	nonce   []byte
 	mfn     string
@@ -140,12 +161,13 @@ type msgBlock struct {
 	cexp    time.Duration
 	ctmr    *time.Timer
 	werr    error
-	loading bool
-	flusher bool
 	dmap    map[uint64]struct{}
 	fch     chan struct{}
 	qch     chan struct{}
 	lchk    [8]byte
+	loading bool
+	flusher bool
+	noSubj  bool
 	closed  bool
 }
 
@@ -215,8 +237,8 @@ const (
 	JetStreamMetaFileKey = "meta.key"
 
 	// AEK key sizes
-	metaKeySize = 72
-	blkKeySize  = 72
+	minMetaKeySize = 64
+	minBlkKeySize  = 64
 
 	// Default stream block size.
 	defaultLargeBlockSize = 8 * 1024 * 1024 // 8MB
@@ -443,8 +465,23 @@ func dynBlkSize(retention RetentionPolicy, maxBytes int64) uint64 {
 	}
 }
 
+func genEncryptionKey(sc StoreCipher, seed []byte) (ek cipher.AEAD, err error) {
+	if sc == ChaCha {
+		ek, err = chacha20poly1305.NewX(seed)
+	} else if sc == AES {
+		block, e := aes.NewCipher(seed)
+		if e != nil {
+			return nil, err
+		}
+		ek, err = cipher.NewGCMWithNonceSize(block, block.BlockSize())
+	} else {
+		err = errUnknownCipher
+	}
+	return ek, err
+}
+
 // Generate an asset encryption key from the context and server PRF.
-func (fs *fileStore) genEncryptionKeys(context string) (aek cipher.AEAD, bek *chacha20.Cipher, seed, encrypted []byte, err error) {
+func (fs *fileStore) genEncryptionKeys(context string) (aek cipher.AEAD, bek cipher.Stream, seed, encrypted []byte, err error) {
 	if fs.prf == nil {
 		return nil, nil, nil, nil, errNoEncryption
 	}
@@ -453,16 +490,22 @@ func (fs *fileStore) genEncryptionKeys(context string) (aek cipher.AEAD, bek *ch
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
-	kek, err := chacha20poly1305.NewX(rb)
+
+	sc := fs.fcfg.Cipher
+
+	kek, err := genEncryptionKey(sc, rb)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
 	// Generate random asset encryption key seed.
-	seed = make([]byte, 32)
-	if n, err := rand.Read(seed); err != nil || n != 32 {
+
+	const seedSize = 32
+	seed = make([]byte, seedSize)
+	if n, err := rand.Read(seed); err != nil || n != seedSize {
 		return nil, nil, nil, nil, err
 	}
-	aek, err = chacha20poly1305.NewX(seed)
+
+	aek, err = genEncryptionKey(sc, seed)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
@@ -470,7 +513,8 @@ func (fs *fileStore) genEncryptionKeys(context string) (aek cipher.AEAD, bek *ch
 	// Generate our nonce. Use same buffer to hold encrypted seed.
 	nonce := make([]byte, kek.NonceSize(), kek.NonceSize()+len(seed)+kek.Overhead())
 	mrand.Read(nonce)
-	bek, err = chacha20.NewUnauthenticatedCipher(seed[:], nonce)
+
+	bek, err = genBlockEncryptionKey(sc, seed[:], nonce)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
@@ -478,6 +522,20 @@ func (fs *fileStore) genEncryptionKeys(context string) (aek cipher.AEAD, bek *ch
 	return aek, bek, seed, kek.Seal(nonce, nonce, seed, nil), nil
 }
 
+// Will generate the block encryption key.
+func genBlockEncryptionKey(sc StoreCipher, seed, nonce []byte) (cipher.Stream, error) {
+	if sc == ChaCha {
+		return chacha20.NewUnauthenticatedCipher(seed, nonce)
+	} else if sc == AES {
+		block, err := aes.NewCipher(seed)
+		if err != nil {
+			return nil, err
+		}
+		return cipher.NewCTR(block, nonce), nil
+	}
+	return nil, errUnknownCipher
+}
+
 // Write out meta and the checksum.
 // Lock should be held.
 func (fs *fileStore) writeStreamMeta() error {
@@ -486,7 +544,6 @@ func (fs *fileStore) writeStreamMeta() error {
 		if err != nil {
 			return err
 		}
-		fs.aek = key
 		keyFile := filepath.Join(fs.fcfg.StoreDir, JetStreamMetaFileKey)
 		if _, err := os.Stat(keyFile); err != nil && !os.IsNotExist(err) {
 			return err
@@ -494,6 +551,8 @@ func (fs *fileStore) writeStreamMeta() error {
 		if err := os.WriteFile(keyFile, encrypted, defaultFilePerms); err != nil {
 			return err
 		}
+		// Set our aek.
+		fs.aek = key
 	}
 
 	meta := filepath.Join(fs.fcfg.StoreDir, JetStreamMetaFile)
@@ -588,6 +647,11 @@ const indexHdrSize = 7*binary.MaxVarintLen64 + hdrLen + checksumSize
 func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, error) {
 	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire}
 
+	// Check if we should be monitoring subjects.
+	if len(fs.cfg.Subjects) == 0 {
+		mb.noSubj = true
+	}
+
 	mdir := filepath.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = filepath.Join(mdir, fi.Name())
 	mb.ifn = filepath.Join(mdir, fmt.Sprintf(indexScan, index))
@@ -611,7 +675,7 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 			}
 			createdKeys = true
 		} else {
-			if len(ekey) != blkKeySize {
+			if len(ekey) < minBlkKeySize {
 				return nil, errBadKeySize
 			}
 			// Recover key encryption key.
@@ -619,20 +683,27 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 			if err != nil {
 				return nil, err
 			}
-			kek, err := chacha20poly1305.NewX(rb)
+
+			sc := fs.fcfg.Cipher
+			kek, err := genEncryptionKey(sc, rb)
 			if err != nil {
 				return nil, err
 			}
 			ns := kek.NonceSize()
 			seed, err := kek.Open(nil, ekey[:ns], ekey[ns:], nil)
 			if err != nil {
-				return nil, err
+				// We may be here on a cipher conversion, so attempt to convert.
+				if err = mb.convertCipher(); err != nil {
+					return nil, err
+				}
+			} else {
+				mb.seed, mb.nonce = seed, ekey[:ns]
 			}
-			mb.seed, mb.nonce = seed, ekey[:ns]
-			if mb.aek, err = chacha20poly1305.NewX(seed); err != nil {
+			mb.aek, err = genEncryptionKey(sc, mb.seed)
+			if err != nil {
 				return nil, err
 			}
-			if mb.bek, err = chacha20.NewUnauthenticatedCipher(seed, ekey[:ns]); err != nil {
+			if mb.bek, err = genBlockEncryptionKey(sc, mb.seed, mb.nonce); err != nil {
 				return nil, err
 			}
 		}
@@ -677,7 +748,7 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 		// Quick sanity check here.
 		// Note this only checks that the message blk file is not newer then this file, or is empty and we expect empty.
 		if (mb.rbytes == 0 && mb.msgs == 0) || bytes.Equal(lchk[:], mb.lchk[:]) {
-			if mb.msgs > 0 && fs.psim != nil {
+			if mb.msgs > 0 && !mb.noSubj && fs.psim != nil {
 				fs.populateGlobalPerSubjectInfo(mb)
 			}
 			fs.addMsgBlock(mb)
@@ -689,7 +760,7 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 	if ld, _ := mb.rebuildState(); ld != nil {
 		fs.rebuildStateLocked(ld)
 	}
-	if mb.msgs > 0 && fs.psim != nil {
+	if mb.msgs > 0 && !mb.noSubj && fs.psim != nil {
 		fs.populateGlobalPerSubjectInfo(mb)
 	}
 
@@ -743,6 +814,75 @@ func (fs *fileStore) rebuildStateLocked(ld *LostStreamData) {
 	}
 }
 
+// Attempt to convert the cipher used for this message block.
+func (mb *msgBlock) convertCipher() error {
+	fs := mb.fs
+	sc := fs.fcfg.Cipher
+
+	var osc StoreCipher
+	switch sc {
+	case ChaCha:
+		osc = AES
+	case AES:
+		osc = ChaCha
+	}
+
+	mdir := filepath.Join(fs.fcfg.StoreDir, msgDir)
+	ekey, err := os.ReadFile(filepath.Join(mdir, fmt.Sprintf(keyScan, mb.index)))
+	if err != nil {
+		return err
+	}
+	if len(ekey) < minBlkKeySize {
+		return errBadKeySize
+	}
+	// Recover key encryption key.
+	rb, err := fs.prf([]byte(fmt.Sprintf("%s:%d", fs.cfg.Name, mb.index)))
+	if err != nil {
+		return err
+	}
+	kek, err := genEncryptionKey(osc, rb)
+	if err != nil {
+		return err
+	}
+	ns := kek.NonceSize()
+	seed, err := kek.Open(nil, ekey[:ns], ekey[ns:], nil)
+	if err != nil {
+		return err
+	}
+	nonce := ekey[:ns]
+
+	bek, err := genBlockEncryptionKey(osc, seed, nonce)
+	if err != nil {
+		return err
+	}
+
+	buf, _ := mb.loadBlock(nil)
+	bek.XORKeyStream(buf, buf)
+	// Make sure we can parse with old cipher and key file.
+	if err = mb.indexCacheBuf(buf); err != nil {
+		return err
+	}
+	// Reset the cache since we just read everything in.
+	mb.cache = nil
+
+	// Generate new keys based on our
+	if err := fs.genEncryptionKeysForBlock(mb); err != nil {
+		// Put the old keyfile back.
+		keyFile := filepath.Join(mdir, fmt.Sprintf(keyScan, mb.index))
+		os.WriteFile(keyFile, ekey, defaultFilePerms)
+		return err
+	}
+	mb.bek.XORKeyStream(buf, buf)
+	if err := os.WriteFile(mb.mfn, buf, defaultFilePerms); err != nil {
+		return err
+	}
+	// If we are here we want to delete other meta, e.g. idx, fss.
+	os.Remove(mb.ifn)
+	os.Remove(mb.sfn)
+
+	return nil
+}
+
 // Convert a plaintext block to encrypted.
 func (mb *msgBlock) convertToEncrypted() error {
 	if mb.bek == nil {
@@ -797,7 +937,7 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 	// Check if we need to decrypt.
 	if mb.bek != nil && len(buf) > 0 {
 		// Recreate to reset counter.
-		mb.bek, err = chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
+		mb.bek, err = genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
 		if err != nil {
 			return nil, err
 		}
@@ -1536,9 +1676,29 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 		return nil
 	}
 
+	start, stop := fs.blks[0], fs.lmb
+	// We can short circuit if not a wildcard using psim for start and stop.
+	if !subjectHasWildcard(subject) {
+		info := fs.psim[subject]
+		if info == nil {
+			return nil
+		}
+		start, stop = fs.bim[info.fblk], fs.bim[info.lblk]
+	}
+
 	fss := make(map[string]SimpleState)
+	var startFound bool
+
 	for _, mb := range fs.blks {
-		mb.mu.RLock()
+		if !startFound {
+			if mb != start {
+				continue
+			}
+			startFound = true
+		}
+
+		mb.mu.Lock()
+		mb.ensurePerSubjectInfoLoaded()
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				oss := fss[subj]
@@ -1551,8 +1711,13 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 				}
 			}
 		}
-		mb.mu.RUnlock()
+		mb.mu.Unlock()
+
+		if mb == stop {
+			break
+		}
 	}
+
 	return fss
 }
 
@@ -1625,8 +1790,10 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 
 	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire}
 
-	// If we only have one subject registered we can optimize filtered lookups here.
-	if len(fs.cfg.Subjects) == 1 {
+	// Optimize lookups if we have zero or one subject.
+	if nsubj := len(fs.cfg.Subjects); nsubj == 0 {
+		mb.noSubj = true
+	} else if nsubj == 1 {
 		mb.sfilter = fs.cfg.Subjects[0]
 	}
 
@@ -1726,9 +1893,6 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts in
 	var psmc uint64
 	psmax := fs.cfg.MaxMsgsPer > 0 && len(subj) > 0
 	if psmax {
-
-		// FIXME(dlc) = psim nil??
-
 		if info, ok := fs.psim[subj]; ok {
 			psmc = info.total
 		}
@@ -2307,7 +2471,7 @@ func (mb *msgBlock) compact() {
 	// Check for encryption.
 	if mb.bek != nil && len(nbuf) > 0 {
 		// Recreate to reset counter.
-		rbek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
+		rbek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
 		if err != nil {
 			return
 		}
@@ -3562,7 +3726,7 @@ checkCache:
 
 	// Check if we need to decrypt.
 	if mb.bek != nil && len(buf) > 0 {
-		bek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
+		bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
 		if err != nil {
 			return err
 		}
@@ -3613,18 +3777,19 @@ func (mb *msgBlock) fetchMsg(seq uint64, sm *StoreMsg) (*StoreMsg, bool, error)
 }
 
 var (
-	errNoCache      = errors.New("no message cache")
-	errBadMsg       = errors.New("malformed or corrupt message")
-	errDeletedMsg   = errors.New("deleted message")
-	errPartialCache = errors.New("partial cache")
-	errNoPending    = errors.New("message block does not have pending data")
-	errNotReadable  = errors.New("storage directory not readable")
-	errCorruptState = errors.New("corrupt state file")
-	errPendingData  = errors.New("pending data still present")
-	errNoEncryption = errors.New("encryption not enabled")
-	errBadKeySize   = errors.New("encryption bad key size")
-	errNoMsgBlk     = errors.New("no message block")
-	errMsgBlkTooBig = errors.New("message block size exceeded int capacity")
+	errNoCache       = errors.New("no message cache")
+	errBadMsg        = errors.New("malformed or corrupt message")
+	errDeletedMsg    = errors.New("deleted message")
+	errPartialCache  = errors.New("partial cache")
+	errNoPending     = errors.New("message block does not have pending data")
+	errNotReadable   = errors.New("storage directory not readable")
+	errCorruptState  = errors.New("corrupt state file")
+	errPendingData   = errors.New("pending data still present")
+	errNoEncryption  = errors.New("encryption not enabled")
+	errBadKeySize    = errors.New("encryption bad key size")
+	errNoMsgBlk      = errors.New("no message block")
+	errMsgBlkTooBig  = errors.New("message block size exceeded int capacity")
+	errUnknownCipher = errors.New("unknown cipher")
 )
 
 // Used for marking messages that have had their checksums checked.
@@ -4616,7 +4781,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 			// Check for encryption.
 			if smb.bek != nil && len(nbuf) > 0 {
 				// Recreate to reset counter.
-				bek, err := chacha20.NewUnauthenticatedCipher(smb.seed, smb.nonce)
+				bek, err := genBlockEncryptionKey(smb.fs.fcfg.Cipher, smb.seed, smb.nonce)
 				if err != nil {
 					goto SKIP
 				}
@@ -4817,7 +4982,7 @@ func (mb *msgBlock) closeAndKeepIndex() {
 
 	// If we are encrypted we should reset our bek counter.
 	if mb.bek != nil {
-		if bek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce); err == nil {
+		if bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce); err == nil {
 			mb.bek = bek
 		}
 	}
@@ -4933,12 +5098,10 @@ func (mb *msgBlock) generatePerSubjectInfo(hasLock bool) error {
 		return nil
 	}
 
-	var shouldExpire bool
 	if mb.cacheNotLoaded() {
 		if err := mb.loadMsgsWithLock(); err != nil {
 			return err
 		}
-		shouldExpire = true
 	}
 
 	// Create new one regardless.
@@ -4967,9 +5130,11 @@ func (mb *msgBlock) generatePerSubjectInfo(hasLock bool) error {
 			}
 		}
 	}
-	if shouldExpire {
-		// Expire this cache before moving on.
-		mb.tryForceExpireCacheLocked()
+
+	if len(mb.fss) > 0 {
+		// Make sure we run the cache expire timer.
+		mb.llts = time.Now().UnixNano()
+		mb.startCacheExpireTimer()
 	}
 	return nil
 }
@@ -5039,9 +5204,9 @@ func (fs *fileStore) populateGlobalPerSubjectInfo(mb *msgBlock) {
 
 // readPerSubjectInfo will attempt to restore the per subject information.
 func (mb *msgBlock) readPerSubjectInfo(hasLock bool) error {
-	// Make sure we run the cache expire timer.
-	mb.llts = time.Now().UnixNano()
-	mb.startCacheExpireTimer()
+	if mb.noSubj {
+		return nil
+	}
 
 	defer func() {
 		if !hasLock {
@@ -5092,6 +5257,12 @@ func (mb *msgBlock) readPerSubjectInfo(hasLock bool) error {
 	}
 	mb.fss = fss
 
+	// Make sure we run the cache expire timer.
+	if len(mb.fss) > 0 {
+		mb.llts = time.Now().UnixNano()
+		mb.startCacheExpireTimer()
+	}
+
 	if !hasLock {
 		mb.mu.Unlock()
 	}
@@ -5366,7 +5537,7 @@ func (fs *fileStore) streamSnapshot(w io.WriteCloser, state *StreamState, includ
 		}
 		// Check for encryption.
 		if mb.bek != nil && len(bbuf) > 0 {
-			rbek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
+			rbek, err := genBlockEncryptionKey(fs.fcfg.Cipher, mb.seed, mb.nonce)
 			if err != nil {
 				mb.mu.Unlock()
 				writeErr(fmt.Sprintf("Could not create encryption key for message block [%d]: %v", mb.index, err))
@@ -5424,7 +5595,7 @@ func (fs *fileStore) streamSnapshot(w io.WriteCloser, state *StreamState, includ
 			writeErr(fmt.Sprintf("Could not encode consumer state for %q: %v", o.name, err))
 			return
 		}
-		odirPre := consumerDir + "/" + o.name
+		odirPre := filepath.Join(consumerDir, o.name)
 		o.mu.Unlock()
 
 		// Write all the consumer files.
@@ -5549,21 +5720,32 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	// Check for encryption.
 	if o.prf != nil {
 		if ekey, err := os.ReadFile(filepath.Join(odir, JetStreamMetaFileKey)); err == nil {
+			if len(ekey) < minBlkKeySize {
+				return nil, errBadKeySize
+			}
 			// Recover key encryption key.
 			rb, err := fs.prf([]byte(fs.cfg.Name + tsep + o.name))
 			if err != nil {
 				return nil, err
 			}
-			kek, err := chacha20poly1305.NewX(rb)
+
+			sc := fs.fcfg.Cipher
+			kek, err := genEncryptionKey(sc, rb)
 			if err != nil {
 				return nil, err
 			}
 			ns := kek.NonceSize()
-			seed, err := kek.Open(nil, ekey[:ns], ekey[ns:], nil)
+			nonce := ekey[:ns]
+			seed, err := kek.Open(nil, nonce, ekey[ns:], nil)
 			if err != nil {
-				return nil, err
+				// We may be here on a cipher conversion, so attempt to convert.
+				if err = o.convertCipher(); err != nil {
+					return nil, err
+				}
+			} else {
+				o.aek, err = genEncryptionKey(sc, seed)
 			}
-			if o.aek, err = chacha20poly1305.NewX(seed); err != nil {
+			if err != nil {
 				return nil, err
 			}
 		}
@@ -5618,6 +5800,63 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	return o, nil
 }
 
+func (o *consumerFileStore) convertCipher() error {
+	fs := o.fs
+	odir := filepath.Join(fs.fcfg.StoreDir, consumerDir, o.name)
+
+	ekey, err := os.ReadFile(filepath.Join(odir, JetStreamMetaFileKey))
+	if err != nil {
+		return err
+	}
+	if len(ekey) < minBlkKeySize {
+		return errBadKeySize
+	}
+	// Recover key encryption key.
+	rb, err := fs.prf([]byte(fs.cfg.Name + tsep + o.name))
+	if err != nil {
+		return err
+	}
+
+	// Do these in reverse since converting.
+	sc := fs.fcfg.Cipher
+	osc := AES
+	if sc == AES {
+		osc = ChaCha
+	}
+	kek, err := genEncryptionKey(osc, rb)
+	if err != nil {
+		return err
+	}
+	ns := kek.NonceSize()
+	nonce := ekey[:ns]
+	seed, err := kek.Open(nil, nonce, ekey[ns:], nil)
+	if err != nil {
+		return err
+	}
+	aek, err := genEncryptionKey(osc, seed)
+	if err != nil {
+		return err
+	}
+	// Now read in and decode our state using the old cipher.
+	buf, err := os.ReadFile(o.ifn)
+	if err != nil {
+		return err
+	}
+	buf, err = aek.Open(nil, buf[:ns], buf[ns:], nil)
+	if err != nil {
+		return err
+	}
+
+	// Since we are here we recovered our old state.
+	// Now write our meta, which will generate the new keys with the new cipher.
+	if err := o.writeConsumerMeta(); err != nil {
+		return err
+	}
+
+	// Now write out or state with the new cipher.
+	return o.writeState(buf)
+}
+
 // Kick flusher for this consumer.
 // Lock should be held.
 func (o *consumerFileStore) kickFlusher() {
diff --git a/server/gateway.go b/server/gateway.go
index 98439ac1..886f204a 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -603,7 +603,7 @@ func (g *srvGateway) generateInfoJSON() {
 	// We could be here when processing a route INFO that has a gateway URL,
 	// but this server is not configured for gateways, so simply ignore here.
 	// The configuration mismatch is reported somewhere else.
-	if !g.enabled {
+	if !g.enabled || g.info == nil {
 		return
 	}
 	g.info.GatewayURLs = g.URLs.getAsStringSlice()
diff --git a/server/jetstream.go b/server/jetstream.go
index 24580346..0eb03aec 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -19,7 +19,6 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"math"
 	"os"
@@ -34,7 +33,6 @@ import (
 	"github.com/nats-io/nats-server/v2/server/sysmem"
 	"github.com/nats-io/nkeys"
 	"github.com/nats-io/nuid"
-	"golang.org/x/crypto/chacha20poly1305"
 )
 
 // JetStreamConfig determines this server's configuration.
@@ -202,10 +200,6 @@ func (s *Server) EnableJetStream(config *JetStreamConfig) error {
 		return err
 	}
 
-	if ek := s.getOpts().JetStreamKey; ek != _EMPTY_ {
-		s.Warnf("JetStream Encryption is Beta")
-	}
-
 	return s.enableJetStream(cfg)
 }
 
@@ -231,9 +225,9 @@ func (s *Server) jsKeyGen(info string) keyGen {
 }
 
 // Decode the encrypted metafile.
-func (s *Server) decryptMeta(ekey, buf []byte, acc, context string) ([]byte, error) {
-	if len(ekey) != metaKeySize {
-		return nil, errors.New("bad encryption key")
+func (s *Server) decryptMeta(sc StoreCipher, ekey, buf []byte, acc, context string) ([]byte, error) {
+	if len(ekey) < minMetaKeySize {
+		return nil, errBadKeySize
 	}
 	prf := s.jsKeyGen(acc)
 	if prf == nil {
@@ -243,7 +237,8 @@ func (s *Server) decryptMeta(ekey, buf []byte, acc, context string) ([]byte, err
 	if err != nil {
 		return nil, err
 	}
-	kek, err := chacha20poly1305.NewX(rb)
+
+	kek, err := genEncryptionKey(sc, rb)
 	if err != nil {
 		return nil, err
 	}
@@ -252,11 +247,10 @@ func (s *Server) decryptMeta(ekey, buf []byte, acc, context string) ([]byte, err
 	if err != nil {
 		return nil, err
 	}
-	aek, err := chacha20poly1305.NewX(seed[:])
+	aek, err := genEncryptionKey(sc, seed)
 	if err != nil {
 		return nil, err
 	}
-	ns = kek.NonceSize()
 	plain, err := aek.Open(nil, buf[:ns], buf[ns:], nil)
 	if err != nil {
 		return nil, err
@@ -371,6 +365,10 @@ func (s *Server) enableJetStream(cfg JetStreamConfig) error {
 	if cfg.Domain != _EMPTY_ {
 		s.Noticef("  Domain:          %s", cfg.Domain)
 	}
+	opts := s.getOpts()
+	if ek := opts.JetStreamKey; ek != _EMPTY_ {
+		s.Noticef("  Encryption:      %s", opts.JetStreamCipher)
+	}
 	s.Noticef("-------------------------------------------")
 
 	// Setup our internal subscriptions.
@@ -1141,6 +1139,11 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 	}
 	var consumers []*ce
 
+	// Remember if we should be encrypted and what cipher we think we should use.
+	encrypted := s.getOpts().JetStreamKey != _EMPTY_
+	plaintext := true
+	sc := s.getOpts().JetStreamCipher
+
 	// Now recover the streams.
 	fis, _ := os.ReadDir(sdir)
 	for _, fi := range fis {
@@ -1177,18 +1180,40 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			continue
 		}
 
+		// Track if we are converting ciphers.
+		var osc StoreCipher
+		var convertingCiphers bool
+
 		// Check if we are encrypted.
-		if key, err := os.ReadFile(filepath.Join(mdir, JetStreamMetaFileKey)); err == nil {
+		keyFile := filepath.Join(mdir, JetStreamMetaFileKey)
+		if key, err := os.ReadFile(keyFile); err == nil {
 			s.Debugf("  Stream metafile is encrypted, reading encrypted keyfile")
-			if len(key) != metaKeySize {
+			if len(key) < minMetaKeySize {
 				s.Warnf("  Bad stream encryption key length of %d", len(key))
 				continue
 			}
 			// Decode the buffer before proceeding.
-			if buf, err = s.decryptMeta(key, buf, a.Name, fi.Name()); err != nil {
-				s.Warnf("  Error decrypting our stream metafile: %v", err)
-				continue
+			nbuf, err := s.decryptMeta(sc, key, buf, a.Name, fi.Name())
+			if err != nil {
+				// See if we are changing ciphers.
+				switch sc {
+				case ChaCha:
+					nbuf, err = s.decryptMeta(AES, key, buf, a.Name, fi.Name())
+					osc, convertingCiphers = AES, true
+				case AES:
+					nbuf, err = s.decryptMeta(ChaCha, key, buf, a.Name, fi.Name())
+					osc, convertingCiphers = ChaCha, true
+				}
+				if err != nil {
+					s.Warnf("  Error decrypting our stream metafile: %v", err)
+					continue
+				}
 			}
+			buf = nbuf
+			plaintext = false
+
+			// Remove the key file to have system regenerate with the new cipher.
+			os.Remove(keyFile)
 		}
 
 		var cfg FileStreamInfo
@@ -1234,6 +1259,15 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 
 		s.Noticef("  Starting restore for stream '%s > %s'", a.Name, cfg.StreamConfig.Name)
 
+		// Log if we are converting from plaintext to encrypted.
+		if encrypted {
+			if plaintext {
+				s.Noticef("  Encrypting stream '%s > %s'", a.Name, cfg.StreamConfig.Name)
+			} else if convertingCiphers {
+				s.Noticef("  Converting from %s to %s for stream '%s > %s'", osc, sc, a.Name, cfg.StreamConfig.Name)
+			}
+		}
+
 		// Add in the stream.
 		mset, err := a.addStream(&cfg.StreamConfig)
 		if err != nil {
@@ -1278,10 +1312,22 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			if key, err := os.ReadFile(filepath.Join(e.odir, ofi.Name(), JetStreamMetaFileKey)); err == nil {
 				s.Debugf("  Consumer metafile is encrypted, reading encrypted keyfile")
 				// Decode the buffer before proceeding.
-				if buf, err = s.decryptMeta(key, buf, a.Name, e.mset.name()+tsep+ofi.Name()); err != nil {
-					s.Warnf("  Error decrypting our consumer metafile: %v", err)
-					continue
+				ctxName := e.mset.name() + tsep + ofi.Name()
+				nbuf, err := s.decryptMeta(sc, key, buf, a.Name, ctxName)
+				if err != nil {
+					// See if we are changing ciphers.
+					switch sc {
+					case ChaCha:
+						nbuf, err = s.decryptMeta(AES, key, buf, a.Name, ctxName)
+					case AES:
+						nbuf, err = s.decryptMeta(ChaCha, key, buf, a.Name, ctxName)
+					}
+					if err != nil {
+						s.Warnf("  Error decrypting our consumer metafile: %v", err)
+						continue
+					}
 				}
+				buf = nbuf
 			}
 
 			var cfg FileConsumerInfo
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 3992618e..d96350f5 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1079,7 +1079,7 @@ func (js *jetStream) metaSnapshot() []byte {
 
 func (js *jetStream) applyMetaSnapshot(buf []byte) error {
 	var wsas []writeableStreamAssignment
-	if len(buf) != 0 {
+	if len(buf) > 0 {
 		jse, err := s2.Decode(nil, buf)
 		if err != nil {
 			return err
@@ -1146,8 +1146,6 @@ func (js *jetStream) applyMetaSnapshot(buf []byte) error {
 		}
 	}
 	isRecovering := js.metaRecovering
-	storeDir := js.config.StoreDir
-	srv := js.srv
 	js.mu.Unlock()
 
 	// Do removals first.
@@ -1185,20 +1183,6 @@ func (js *jetStream) applyMetaSnapshot(buf []byte) error {
 		js.processConsumerAssignment(ca)
 	}
 
-	// Delete directories not part of snapshot
-	sysAcc := srv.SystemAccount().GetName()
-	if de, err := os.ReadDir(storeDir); err == nil {
-		for _, entry := range de {
-			acc := entry.Name()
-			if acc == sysAcc {
-				continue
-			}
-			if _, found := streams[acc]; !found {
-				os.RemoveAll(filepath.Join(storeDir, acc))
-			}
-		}
-	}
-
 	return nil
 }
 
@@ -1416,7 +1400,9 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, rm *recoveryRemovals) (b
 					key := sa.Client.Account + ":" + sa.Config.Name
 					delete(rm.streams, key)
 				}
-				didRemove = js.processStreamAssignment(sa)
+				if js.processStreamAssignment(sa) {
+					didRemove = true
+				}
 			case removeStreamOp:
 				sa, err := decodeStreamAssignment(buf[1:])
 				if err != nil {
@@ -5090,7 +5076,34 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 
 	// Check if this is a move request, but no cancellation, and we are already moving this stream.
 	if isMoveRequest && !isMoveCancel && osa.Config.Replicas != len(rg.Peers) {
-		resp.Error = NewJSStreamMoveInProgressError()
+		// obtain stats to include in error message
+		msg := _EMPTY_
+		if s.allPeersOffline(rg) {
+			msg = fmt.Sprintf("all %d peers offline", len(rg.Peers))
+		} else {
+			// Need to release js lock.
+			js.mu.Unlock()
+			if si, err := s.sysRequest(&StreamInfo{}, clusterStreamInfoT, ci.serviceAccount(), cfg.Name); err != nil {
+				msg = fmt.Sprintf("error retrieving info: %s", err.Error())
+			} else if si := si.(*StreamInfo); si != nil {
+				currentCount := 0
+				if si.Cluster.Leader != _EMPTY_ {
+					currentCount++
+				}
+				combinedLag := uint64(0)
+				for _, r := range si.Cluster.Replicas {
+					if r.Current {
+						currentCount++
+					}
+					combinedLag += r.Lag
+				}
+				msg = fmt.Sprintf("total peers: %d, current peers: %d, combined lag: %d",
+					len(rg.Peers), currentCount, combinedLag)
+			}
+			// Re-acquire here.
+			js.mu.Lock()
+		}
+		resp.Error = NewJSStreamMoveInProgressError(msg)
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 		return
 	}
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index ce9f62ac..91419f93 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -311,8 +311,8 @@ const (
 	// JSStreamMoveAndScaleErr can not move and scale a stream in a single update
 	JSStreamMoveAndScaleErr ErrorIdentifier = 10123
 
-	// JSStreamMoveInProgress stream move already in progress
-	JSStreamMoveInProgress ErrorIdentifier = 10124
+	// JSStreamMoveInProgressF stream move already in progress: {msg}
+	JSStreamMoveInProgressF ErrorIdentifier = 10124
 
 	// JSStreamMoveNotInProgress stream move not in progress
 	JSStreamMoveNotInProgress ErrorIdentifier = 10129
@@ -497,7 +497,7 @@ var (
 		JSStreamMirrorNotUpdatableErr:              {Code: 400, ErrCode: 10055, Description: "stream mirror configuration can not be updated"},
 		JSStreamMismatchErr:                        {Code: 400, ErrCode: 10056, Description: "stream name in subject does not match request"},
 		JSStreamMoveAndScaleErr:                    {Code: 400, ErrCode: 10123, Description: "can not move and scale a stream in a single update"},
-		JSStreamMoveInProgress:                     {Code: 400, ErrCode: 10124, Description: "stream move already in progress"},
+		JSStreamMoveInProgressF:                    {Code: 400, ErrCode: 10124, Description: "stream move already in progress: {msg}"},
 		JSStreamMoveNotInProgress:                  {Code: 400, ErrCode: 10129, Description: "stream move not in progress"},
 		JSStreamMsgDeleteFailedF:                   {Code: 500, ErrCode: 10057, Description: "{err}"},
 		JSStreamNameContainsPathSeparatorsErr:      {Code: 400, ErrCode: 10128, Description: "Stream name can not contain path separators"},
@@ -1707,14 +1707,20 @@ func NewJSStreamMoveAndScaleError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSStreamMoveAndScaleErr]
 }
 
-// NewJSStreamMoveInProgressError creates a new JSStreamMoveInProgress error: "stream move already in progress"
-func NewJSStreamMoveInProgressError(opts ...ErrorOption) *ApiError {
+// NewJSStreamMoveInProgressError creates a new JSStreamMoveInProgressF error: "stream move already in progress: {msg}"
+func NewJSStreamMoveInProgressError(msg interface{}, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
 	if ae, ok := eopts.err.(*ApiError); ok {
 		return ae
 	}
 
-	return ApiErrors[JSStreamMoveInProgress]
+	e := ApiErrors[JSStreamMoveInProgressF]
+	args := e.toReplacerArgs([]interface{}{"{msg}", msg})
+	return &ApiError{
+		Code:        e.Code,
+		ErrCode:     e.ErrCode,
+		Description: strings.NewReplacer(args...).Replace(e.Description),
+	}
 }
 
 // NewJSStreamMoveNotInProgressError creates a new JSStreamMoveNotInProgress error: "stream move not in progress"
diff --git a/server/monitor.go b/server/monitor.go
index 8dc872b5..78e88ced 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1777,7 +1777,7 @@ func (s *Server) Gatewayz(opts *GatewayzOptions) (*Gatewayz, error) {
 	now := time.Now().UTC()
 	gw := s.gateway
 	gw.RLock()
-	if !gw.enabled {
+	if !gw.enabled || gw.info == nil {
 		gw.RUnlock()
 		gwz := &Gatewayz{
 			ID:               srvID,
diff --git a/server/opts.go b/server/opts.go
index fe2ab2b4..ecfa1d5e 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -252,6 +252,7 @@ type Options struct {
 	JetStreamDomain       string        `json:"-"`
 	JetStreamExtHint      string        `json:"-"`
 	JetStreamKey          string        `json:"-"`
+	JetStreamCipher       StoreCipher   `json:"-"`
 	JetStreamUniqueTag    string
 	JetStreamLimits       JSLimitOpts
 	StoreDir              string            `json:"-"`
@@ -1930,6 +1931,15 @@ func parseJetStream(v interface{}, opts *Options, errors *[]error, warnings *[]e
 				doEnable = mv.(bool)
 			case "key", "ek", "encryption_key":
 				opts.JetStreamKey = mv.(string)
+			case "cipher":
+				switch strings.ToLower(mv.(string)) {
+				case "chacha", "chachapoly":
+					opts.JetStreamCipher = ChaCha
+				case "aes":
+					opts.JetStreamCipher = AES
+				default:
+					return &configErr{tk, fmt.Sprintf("Unknown cipher type: %q", mv)}
+				}
 			case "extension_hint":
 				opts.JetStreamExtHint = mv.(string)
 			case "limits":
diff --git a/server/raft.go b/server/raft.go
index 419f580a..f43083b1 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -569,6 +569,27 @@ func (s *Server) reloadDebugRaftNodes() {
 	s.rnMu.RUnlock()
 }
 
+func (s *Server) stepdownRaftNodes() {
+	if s == nil {
+		return
+	}
+	var nodes []RaftNode
+	s.rnMu.RLock()
+	if len(s.raftNodes) > 0 {
+		s.Debugf("Stepping down all leader raft nodes")
+	}
+	for _, n := range s.raftNodes {
+		if n.Leader() {
+			nodes = append(nodes, n)
+		}
+	}
+	s.rnMu.RUnlock()
+
+	for _, node := range nodes {
+		node.StepDown()
+	}
+}
+
 func (s *Server) shutdownRaftNodes() {
 	if s == nil {
 		return
@@ -584,9 +605,6 @@ func (s *Server) shutdownRaftNodes() {
 	s.rnMu.RUnlock()
 
 	for _, node := range nodes {
-		if node.Leader() {
-			node.StepDown()
-		}
 		node.Stop()
 	}
 }
@@ -973,7 +991,7 @@ func (n *raft) InstallSnapshot(data []byte) error {
 	// Remember our latest snapshot file.
 	n.snapfile = sfile
 
-	if _, err := n.wal.Compact(snap.lastIndex); err != nil {
+	if _, err := n.wal.Compact(snap.lastIndex + 1); err != nil {
 		n.Unlock()
 		n.setWriteErr(err)
 		return err
diff --git a/server/reload.go b/server/reload.go
index 372414e1..0b23b298 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -945,7 +945,7 @@ func imposeOrder(value interface{}) error {
 		sort.Strings(value.AllowedOrigins)
 	case string, bool, uint8, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
-		*OCSPConfig, map[string]string, JSLimitOpts:
+		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher:
 		// explicitly skipped types
 	default:
 		// this will fail during unit tests
@@ -976,7 +976,7 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 		field := oldConfig.Type().Field(i)
 		// field.PkgPath is empty for exported fields, and is not for unexported ones.
 		// We skip the unexported fields.
-		if field.PkgPath != "" {
+		if field.PkgPath != _EMPTY_ {
 			continue
 		}
 		var (
diff --git a/server/server.go b/server/server.go
index 83525ac0..3515daf4 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1872,8 +1872,8 @@ func (s *Server) Shutdown() {
 	if s == nil {
 		return
 	}
-	// Transfer off any raft nodes that we are a leader by shutting them all down.
-	s.shutdownRaftNodes()
+	// Transfer off any raft nodes that we are a leader by stepping them down.
+	s.stepdownRaftNodes()
 
 	// This is for clustered JetStream and ephemeral consumers.
 	// No-op if not clustered or not running JetStream.
@@ -1911,6 +1911,9 @@ func (s *Server) Shutdown() {
 	// Now check jetstream.
 	s.shutdownJetStream()
 
+	// Now shutdown the nodes
+	s.shutdownRaftNodes()
+
 	s.mu.Lock()
 	conns := make(map[uint64]*client)
 
diff --git a/server/stream.go b/server/stream.go
index aa903907..3be5616d 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -402,7 +402,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	// Check for overlapping subjects. These are not allowed for now.
 	if jsa.subjectsOverlap(cfg.Subjects) {
 		jsa.mu.Unlock()
-		return nil, fmt.Errorf("subjects overlap with an existing stream")
+		return nil, NewJSStreamSubjectOverlapError()
 	}
 
 	if !hasTier {
@@ -1439,12 +1439,27 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 					si := &sourceInfo{name: s.Name, iname: s.iname}
 					mset.sources[s.iname] = si
 					mset.setStartingSequenceForSource(s.iname)
-					mset.setSourceConsumer(s.iname, si.sseq+1)
+					mset.setSourceConsumer(s.iname, si.sseq+1, time.Time{})
 				} else if oFilter != s.FilterSubject {
 					if si, ok := mset.sources[s.iname]; ok {
-						// si.sseq is the last message we received
-						// if upstream has more messages, that we used to filter, now we get now
-						mset.setSourceConsumer(s.iname, si.sseq+1)
+						filterOverlap := true
+						if oFilter != _EMPTY_ && s.FilterSubject != _EMPTY_ {
+							newFilter := strings.Split(s.FilterSubject, tsep)
+							oldFilter := strings.Split(oFilter, tsep)
+							if !isSubsetMatchTokenized(oldFilter, newFilter) &&
+								!isSubsetMatchTokenized(newFilter, oldFilter) {
+								filterOverlap = false
+							}
+						}
+						if filterOverlap {
+							// si.sseq is the last message we received
+							// if upstream has more messages (with a bigger sequence number)
+							// that we used to filter, now we get them
+							mset.setSourceConsumer(s.iname, si.sseq+1, time.Time{})
+						} else {
+							// since the filter has no overlap at all, we will request messages starting now
+							mset.setSourceConsumer(s.iname, si.sseq+1, time.Now())
+						}
 					}
 				}
 				delete(current, s.iname)
@@ -2311,7 +2326,7 @@ func (mset *stream) retrySourceConsumerAtSeq(iname string, seq uint64) {
 	s.Debugf("Retrying source consumer for '%s > %s'", mset.acc.Name, mset.cfg.Name)
 
 	// setSourceConsumer will check that the source is still configured.
-	mset.setSourceConsumer(iname, seq)
+	mset.setSourceConsumer(iname, seq, time.Time{})
 }
 
 // Lock should be held.
@@ -2354,7 +2369,7 @@ const sourceConsumerRetryThreshold = 2 * time.Second
 // without tripping the sourceConsumerRetryThreshold.
 //
 // Lock held on entry
-func (mset *stream) scheduleSetSourceConsumerRetryAsap(si *sourceInfo, seq uint64) {
+func (mset *stream) scheduleSetSourceConsumerRetryAsap(si *sourceInfo, seq uint64, startTime time.Time) {
 	// We are trying to figure out how soon we can retry. setSourceConsumer will reject
 	// a retry if last was done less than "sourceConsumerRetryThreshold" ago.
 	next := sourceConsumerRetryThreshold - time.Since(si.lreq)
@@ -2365,22 +2380,22 @@ func (mset *stream) scheduleSetSourceConsumerRetryAsap(si *sourceInfo, seq uint6
 	// To make *sure* that the next request will not fail, add a bit of buffer
 	// and some randomness.
 	next += time.Duration(rand.Intn(50)) + 10*time.Millisecond
-	mset.scheduleSetSourceConsumerRetry(si.iname, seq, next)
+	mset.scheduleSetSourceConsumerRetry(si.iname, seq, next, startTime)
 }
 
 // Simply schedules setSourceConsumer at the given delay.
 //
 // Does not require lock
-func (mset *stream) scheduleSetSourceConsumerRetry(iname string, seq uint64, delay time.Duration) {
+func (mset *stream) scheduleSetSourceConsumerRetry(iname string, seq uint64, delay time.Duration, startTime time.Time) {
 	time.AfterFunc(delay, func() {
 		mset.mu.Lock()
-		mset.setSourceConsumer(iname, seq)
+		mset.setSourceConsumer(iname, seq, startTime)
 		mset.mu.Unlock()
 	})
 }
 
 // Lock should be held.
-func (mset *stream) setSourceConsumer(iname string, seq uint64) {
+func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.Time) {
 	si := mset.sources[iname]
 	if si == nil {
 		return
@@ -2396,7 +2411,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 	// We want to throttle here in terms of how fast we request new consumers,
 	// or if the previous is still in progress.
 	if last := time.Since(si.lreq); last < sourceConsumerRetryThreshold || si.sip {
-		mset.scheduleSetSourceConsumerRetryAsap(si, seq)
+		mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
 		return
 	}
 	si.lreq = time.Now()
@@ -2425,7 +2440,10 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 	}
 
 	// If starting, check any configs.
-	if seq <= 1 {
+	if !startTime.IsZero() && seq > 1 {
+		req.Config.OptStartTime = &startTime
+		req.Config.DeliverPolicy = DeliverByStartTime
+	} else if seq <= 1 {
 		if ssi.OptStartSeq > 0 {
 			req.Config.OptStartSeq = ssi.OptStartSeq
 			req.Config.DeliverPolicy = DeliverByStartSequence
@@ -2460,7 +2478,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 	})
 	if err != nil {
 		si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
-		mset.scheduleSetSourceConsumerRetryAsap(si, seq)
+		mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
 		return
 	}
 
@@ -2486,7 +2504,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 	if err != nil {
 		si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
 		mset.unsubscribeUnlocked(crSub)
-		mset.scheduleSetSourceConsumerRetryAsap(si, seq)
+		mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
 		return
 	}
 	si.err = nil
@@ -2506,7 +2524,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 				si.sip = false
 				// If we need to retry, schedule now
 				if retry {
-					mset.scheduleSetSourceConsumerRetryAsap(si, seq)
+					mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
 				}
 			}
 			mset.mu.Unlock()
@@ -2606,7 +2624,7 @@ func (mset *stream) processSourceMsgs(si *sourceInfo, ready *sync.WaitGroup) {
 				mset.mu.Lock()
 				// We don't need to schedule here, we are going to simply
 				// call setSourceConsumer with the current state+1.
-				mset.setSourceConsumer(iname, si.sseq+1)
+				mset.setSourceConsumer(iname, si.sseq+1, time.Time{})
 				mset.mu.Unlock()
 			}
 		}
@@ -2934,7 +2952,7 @@ func (mset *stream) setupSourceConsumers() error {
 	// Setup our consumers at the proper starting position.
 	for _, ssi := range mset.cfg.Sources {
 		if si := mset.sources[ssi.iname]; si != nil {
-			mset.setSourceConsumer(ssi.iname, si.sseq+1)
+			mset.setSourceConsumer(ssi.iname, si.sseq+1, time.Time{})
 		}
 	}
 
@@ -3168,7 +3186,12 @@ func (mset *stream) setupStore(fsCfg *FileStoreConfig) error {
 		mset.store = ms
 	case FileStorage:
 		s := mset.srv
-		fs, err := newFileStoreWithCreated(*fsCfg, mset.cfg, mset.created, s.jsKeyGen(mset.acc.Name))
+		prf := s.jsKeyGen(mset.acc.Name)
+		if prf != nil {
+			// We are encrypted here, fill in correct cipher selection.
+			fsCfg.Cipher = s.getOpts().JetStreamCipher
+		}
+		fs, err := newFileStoreWithCreated(*fsCfg, mset.cfg, mset.created, prf)
 		if err != nil {
 			mset.mu.Unlock()
 			return err


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 1aa99599..82da85e4 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -980,7 +980,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 
 	fs, err := newFileStoreWithCreated(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 350},
-		StreamConfig{Name: "zzz", Storage: FileStorage},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
 		time.Now(),
 		prf,
 	)
@@ -1037,7 +1037,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 
 	fs, err = newFileStoreWithCreated(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 350},
-		StreamConfig{Name: "zzz", Storage: FileStorage},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
 		time.Now(),
 		prf,
 	)
@@ -1059,7 +1059,10 @@ func TestFileStoreRemovePartialRecovery(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
-	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1088,7 +1091,10 @@ func TestFileStoreRemovePartialRecovery(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1104,7 +1110,10 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
-	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1145,7 +1154,10 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1765,7 +1777,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
-		StreamConfig{Name: "zzz", Storage: FileStorage},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
 	)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -1848,7 +1860,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 
 		fsr, err := newFileStore(
 			FileStoreConfig{StoreDir: rstoreDir},
-			StreamConfig{Name: "zzz", Storage: FileStorage},
+			StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
 		)
 		if err != nil {
 			t.Fatalf("Error restoring from snapshot: %v", err)
@@ -4099,7 +4111,7 @@ func TestFileStoreEncryptedKeepIndexNeedBekResetBug(t *testing.T) {
 		fs.StoreMsg(subj, nil, msg)
 	}
 
-	// What to go to 0.
+	// Want to go to 0.
 	// This will leave the marker.
 	checkFor(t, time.Second, ttl, func() error {
 		if state := fs.State(); state.Msgs != 0 {
@@ -4149,7 +4161,7 @@ func TestFileStoreExpireSubjectMeta(t *testing.T) {
 
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 1024, CacheExpire: time.Second},
-		StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgsPer: 1},
+		StreamConfig{Name: "zzz", Subjects: []string{"kv.>"}, Storage: FileStorage, MaxMsgsPer: 1},
 	)
 	require_NoError(t, err)
 	defer fs.Stop()
@@ -4173,7 +4185,7 @@ func TestFileStoreExpireSubjectMeta(t *testing.T) {
 	fs.Stop()
 	fs, err = newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 1024, CacheExpire: time.Second},
-		StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgsPer: 1},
+		StreamConfig{Name: "zzz", Subjects: []string{"kv.>"}, Storage: FileStorage, MaxMsgsPer: 1},
 	)
 	require_NoError(t, err)
 	defer fs.Stop()
@@ -4241,3 +4253,173 @@ func TestFileStoreMaxMsgsPerSubject(t *testing.T) {
 		t.Fatalf("Expected 34 blocks, got %d", nb)
 	}
 }
+
+func TestFileStoreSubjectStateCacheExpiration(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 32, CacheExpire: time.Second},
+		StreamConfig{Name: "zzz", Subjects: []string{"kv.>"}, Storage: FileStorage, MaxMsgsPer: 2},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	for i := 1; i <= 100; i++ {
+		subj := fmt.Sprintf("kv.foo.%d", i)
+		_, _, err := fs.StoreMsg(subj, nil, []byte("value"))
+		require_NoError(t, err)
+	}
+	for i := 1; i <= 100; i++ {
+		subj := fmt.Sprintf("kv.bar.%d", i)
+		_, _, err := fs.StoreMsg(subj, nil, []byte("value"))
+		require_NoError(t, err)
+	}
+
+	// Make sure we clear mb fss meta before asking for SubjectState.
+	checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+		if _, hasAnyFSS := fs.reportMeta(); hasAnyFSS {
+			return fmt.Errorf("Still have mb fss state")
+		}
+		return nil
+	})
+
+	if fss := fs.SubjectsState("kv.bar.>"); len(fss) != 100 {
+		t.Fatalf("Expected 100 entries but got %d", len(fss))
+	}
+
+	fss := fs.SubjectsState("kv.bar.99")
+	if len(fss) != 1 {
+		t.Fatalf("Expected 1 entry but got %d", len(fss))
+	}
+	expected := SimpleState{Msgs: 1, First: 199, Last: 199}
+	if ss := fss["kv.bar.99"]; ss != expected {
+		t.Fatalf("Bad subject state, expected %+v but got %+v", expected, ss)
+	}
+
+	// Now add one to end and check as well for non-wildcard.
+	_, _, err = fs.StoreMsg("kv.foo.1", nil, []byte("value22"))
+	require_NoError(t, err)
+
+	if state := fs.State(); state.Msgs != 201 {
+		t.Fatalf("Expected 201 msgs but got %+v", state)
+	}
+
+	fss = fs.SubjectsState("kv.foo.1")
+	if len(fss) != 1 {
+		t.Fatalf("Expected 1 entry but got %d", len(fss))
+	}
+	expected = SimpleState{Msgs: 2, First: 1, Last: 201}
+	if ss := fss["kv.foo.1"]; ss != expected {
+		t.Fatalf("Bad subject state, expected %+v but got %+v", expected, ss)
+	}
+}
+
+func TestFileStoreEncryptedAES(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer os.RemoveAll(storeDir)
+
+	prf := func(context []byte) ([]byte, error) {
+		h := hmac.New(sha256.New, []byte("dlc22"))
+		if _, err := h.Write(context); err != nil {
+			return nil, err
+		}
+		return h.Sum(nil), nil
+	}
+
+	fs, err := newFileStoreWithCreated(
+		FileStoreConfig{StoreDir: storeDir, Cipher: AES},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+		time.Now(),
+		prf,
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("aes ftw")
+	for i := 0; i < 50; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+
+	o, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+	require_NoError(t, err)
+
+	state := &ConsumerState{}
+	state.Delivered.Consumer = 22
+	state.Delivered.Stream = 22
+	state.AckFloor.Consumer = 11
+	state.AckFloor.Stream = 11
+	err = o.Update(state)
+	require_NoError(t, err)
+
+	fs.Stop()
+
+	fs, err = newFileStoreWithCreated(
+		FileStoreConfig{StoreDir: storeDir, Cipher: AES},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+		time.Now(),
+		prf,
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Now make sure we can read.
+	var smv StoreMsg
+	sm, err := fs.LoadMsg(10, &smv)
+	require_NoError(t, err)
+	require_True(t, string(sm.msg) == "aes ftw")
+
+	o, err = fs.ConsumerStore("o22", &ConsumerConfig{})
+	require_NoError(t, err)
+	rstate, err := o.State()
+	require_NoError(t, err)
+
+	if rstate.Delivered != state.Delivered || rstate.AckFloor != state.AckFloor {
+		t.Fatalf("Bad recovered consumer state, expected %+v got %+v", state, rstate)
+	}
+}
+
+// Make sure we do not go through block loads when we know no subjects will exists, e.g. raft.
+func TestFileStoreNoFSSWhenNoSubjects(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	n, msg := 100, []byte("raft state")
+	for i := 0; i < n; i++ {
+		_, _, err := fs.StoreMsg(_EMPTY_, nil, msg)
+		require_NoError(t, err)
+	}
+
+	state := fs.State()
+	require_True(t, state.Msgs == uint64(n))
+
+	fs.Stop()
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Make sure we did not load the block trying to generate fss.
+	fs.mu.RLock()
+	mb := fs.blks[0]
+	fs.mu.RUnlock()
+
+	mb.mu.Lock()
+	defer mb.mu.Unlock()
+
+	if mb.cloads > 0 {
+		t.Fatalf("Expected no cache loads but got %d", mb.cloads)
+	}
+	if mb.fss != nil {
+		t.Fatalf("Expected fss to be nil")
+	}
+}
diff --git a/server/gateway_test.go b/server/gateway_test.go
index bbde745c..b8f9c5ab 100644
--- a/server/gateway_test.go
+++ b/server/gateway_test.go
@@ -6749,3 +6749,34 @@ func TestGatewayDuplicateServerName(t *testing.T) {
 	// to cluster "A"
 	checkForDupError(scl.errCh)
 }
+
+func TestGatewayNoPanicOnStartupWithMonitoring(t *testing.T) {
+	o := testDefaultOptionsForGateway("B")
+	o.HTTPHost = "127.0.0.1"
+	o.HTTPPort = 8888
+	s, err := NewServer(o)
+	require_NoError(t, err)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		time.Sleep(50 * time.Millisecond)
+		s.Start()
+		s.WaitForShutdown()
+	}()
+
+	for {
+		g, err := s.Gatewayz(nil)
+		if err != nil {
+			continue
+		}
+		if g.Port != 0 && g.Port != s.GatewayAddr().Port {
+			t.Fatalf("Unexpected port: %v vs %v", g.Port, s.GatewayAddr().Port)
+		}
+		break
+	}
+	s.Shutdown()
+	wg.Wait()
+}
diff --git a/server/jetstream_chaos_consumer_test.go b/server/jetstream_chaos_consumer_test.go
new file mode 100644
index 00000000..bd21104b
--- /dev/null
+++ b/server/jetstream_chaos_consumer_test.go
@@ -0,0 +1,615 @@
+// Copyright 2022 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build js_chaos_tests
+// +build js_chaos_tests
+
+package server
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+const (
+	chaosConsumerTestsClusterName = "CONSUMERS_CHAOS_TEST"
+	chaosConsumerTestsStreamName  = "CONSUMER_CHAOS_TEST_STREAM"
+	chaosConsumerTestsSubject     = "foo"
+	chaosConsumerTestsDebug       = false
+)
+
+// Creates stream and fills it with the given number of messages.
+// Each message is the string representation of the stream sequence number,
+// e.g. the first message (seqno: 1) contains data "1".
+// This allows consumers to verify the content of each message without tracking additional state
+func createStreamForConsumerChaosTest(t *testing.T, c *cluster, replicas, numMessages int) {
+	t.Helper()
+
+	const publishBatchSize = 1_000
+
+	pubNc, pubJs := jsClientConnectCluster(t, c)
+	defer pubNc.Close()
+
+	_, err := pubJs.AddStream(&nats.StreamConfig{
+		Name:     chaosConsumerTestsStreamName,
+		Subjects: []string{chaosConsumerTestsSubject},
+		Replicas: replicas,
+	})
+	if err != nil {
+		t.Fatalf("Error creating stream: %v", err)
+	}
+
+	ackFutures := make([]nats.PubAckFuture, 0, publishBatchSize)
+
+	for i := 1; i <= numMessages; i++ {
+		message := []byte(fmt.Sprintf("%d", i))
+		pubAckFuture, err := pubJs.PublishAsync(chaosConsumerTestsSubject, message, nats.ExpectLastSequence(uint64(i-1)))
+		if err != nil {
+			t.Fatalf("Publish error: %s", err)
+		}
+		ackFutures = append(ackFutures, pubAckFuture)
+
+		if (i > 0 && i%publishBatchSize == 0) || i == numMessages {
+			select {
+			case <-pubJs.PublishAsyncComplete():
+				for _, pubAckFuture := range ackFutures {
+					select {
+					case <-pubAckFuture.Ok():
+						// Noop
+					case pubAckErr := <-pubAckFuture.Err():
+						t.Fatalf("Error publishing: %s", pubAckErr)
+					case <-time.After(30 * time.Second):
+						t.Fatalf("Timeout verifying pubAck for message: %s", pubAckFuture.Msg().Data)
+					}
+				}
+				ackFutures = make([]nats.PubAckFuture, 0, publishBatchSize)
+				t.Logf("Published %d/%d messages", i, numMessages)
+
+			case <-time.After(30 * time.Second):
+				t.Fatalf("Publish timed out")
+			}
+		}
+	}
+}
+
+// Verify ordered delivery despite cluster-wide outages
+func TestJetStreamChaosConsumerOrdered(t *testing.T) {
+
+	const numMessages = 30_000
+	const maxRetries = 100
+	const retryDelay = 500 * time.Millisecond
+	const fetchTimeout = 250 * time.Millisecond
+	const clusterSize = 3
+	const replicas = 3
+
+	c := createJetStreamClusterExplicit(t, chaosConsumerTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createStreamForConsumerChaosTest(t, c, replicas, numMessages)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: clusterSize, // Whole cluster outage
+			maxDownServers: clusterSize,
+			pause:          1 * time.Second,
+		},
+	)
+
+	subNc, subJs := jsClientConnectCluster(t, c)
+	defer subNc.Close()
+
+	sub, err := subJs.SubscribeSync(
+		chaosConsumerTestsSubject,
+		nats.OrderedConsumer(),
+	)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	if chaosConsumerTestsDebug {
+		t.Logf("Initial subscription: %s", toIndentedJsonString(sub))
+	}
+
+	chaos.start()
+	defer chaos.stop()
+
+	for i := 1; i <= numMessages; i++ {
+		var msg *nats.Msg
+		var nextMsgErr error
+		var expectedMsgData = []byte(fmt.Sprintf("%d", i))
+
+	nextMsgRetryLoop:
+		for r := 0; r <= maxRetries; r++ {
+			msg, nextMsgErr = sub.NextMsg(fetchTimeout)
+			if nextMsgErr == nil {
+				break nextMsgRetryLoop
+			} else if r == maxRetries {
+				t.Fatalf("Exceeded max retries for NextMsg")
+			} else if nextMsgErr == nats.ErrBadSubscription {
+				t.Fatalf("Subscription is invalid: %s", toIndentedJsonString(sub))
+			} else {
+				time.Sleep(retryDelay)
+			}
+		}
+
+		metadata, err := msg.Metadata()
+		if err != nil {
+			t.Fatalf("Failed to get message metadata: %v", err)
+		}
+
+		if metadata.Sequence.Stream != uint64(i) {
+			t.Fatalf("Expecting stream sequence %d, got %d instead", i, metadata.Sequence.Stream)
+		}
+
+		if !bytes.Equal(msg.Data, expectedMsgData) {
+			t.Fatalf("Expecting message %s, got %s instead", expectedMsgData, msg.Data)
+		}
+
+		// Simulate application processing (and gives the monkey some time to brew chaos)
+		time.Sleep(10 * time.Millisecond)
+
+		if i%1000 == 0 {
+			t.Logf("Consumed %d/%d", i, numMessages)
+		}
+	}
+}
+
+// Verify ordered delivery despite cluster-wide outages
+func TestJetStreamChaosConsumerAsync(t *testing.T) {
+
+	const numMessages = 30_000
+	const timeout = 30 * time.Second // No (new) messages for 30s => terminate
+	const maxRetries = 25
+	const retryDelay = 500 * time.Millisecond
+	const clusterSize = 3
+	const replicas = 3
+
+	c := createJetStreamClusterExplicit(t, chaosConsumerTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createStreamForConsumerChaosTest(t, c, replicas, numMessages)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: clusterSize,
+			maxDownServers: clusterSize,
+			pause:          2 * time.Second,
+		},
+	)
+
+	subNc, subJs := jsClientConnectCluster(t, c)
+	defer subNc.Close()
+
+	timeoutTimer := time.NewTimer(timeout)
+	deliveryCount := uint64(0)
+	received := NewBitset(numMessages)
+
+	handleMsg := func(msg *nats.Msg) {
+		deliveryCount += 1
+
+		metadata, err := msg.Metadata()
+		if err != nil {
+			t.Fatalf("Failed to get message metadata: %v", err)
+		}
+		seq := metadata.Sequence.Stream
+
+		var expectedMsgData = []byte(fmt.Sprintf("%d", seq))
+		if !bytes.Equal(msg.Data, expectedMsgData) {
+			t.Fatalf("Expecting message content '%s', got '%s' instead", expectedMsgData, msg.Data)
+		}
+
+		isDupe := received.get(seq - 1)
+
+		if isDupe {
+			if chaosConsumerTestsDebug {
+				t.Logf("Duplicate message delivery, seq: %d", seq)
+			}
+			return
+		}
+
+		// Mark this sequence as received
+		received.set(seq-1, true)
+		if received.count() < numMessages {
+			// Reset timeout
+			timeoutTimer.Reset(timeout)
+		} else {
+			// All received, speed up the shutdown
+			timeoutTimer.Reset(1 * time.Second)
+		}
+
+		if received.count()%1000 == 0 {
+			t.Logf("Consumed %d/%d", received.count(), numMessages)
+		}
+
+		// Simulate application processing (and gives the monkey some time to brew chaos)
+		time.Sleep(10 * time.Millisecond)
+
+	ackRetryLoop:
+		for i := 0; i <= maxRetries; i++ {
+			ackErr := msg.Ack()
+			if ackErr == nil {
+				break ackRetryLoop
+			} else if i == maxRetries {
+				t.Fatalf("Failed to ACK message %d (retried %d times)", seq, maxRetries)
+			} else {
+				time.Sleep(retryDelay)
+			}
+		}
+	}
+
+	subOpts := []nats.SubOpt{}
+	sub, err := subJs.Subscribe(chaosConsumerTestsSubject, handleMsg, subOpts...)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	chaos.start()
+	defer chaos.stop()
+
+	// Wait for long enough silence.
+	// Either a stall, or all messages received
+	<-timeoutTimer.C
+
+	// Shut down consumer
+	sub.Unsubscribe()
+
+	uniqueDeliveredCount := received.count()
+
+	t.Logf(
+		"Delivered %d/%d messages %d duplicate deliveries",
+		uniqueDeliveredCount,
+		numMessages,
+		deliveryCount-uniqueDeliveredCount,
+	)
+
+	if uniqueDeliveredCount != numMessages {
+		t.Fatalf("No new message delivered in the last %s, %d/%d messages never delivered", timeout, numMessages-uniqueDeliveredCount, numMessages)
+	}
+}
+
+// Verify durable consumer retains state despite cluster-wide outages
+// The consumer connection is also periodically closed, and the consumer 'resumes' on a different one
+func TestJetStreamChaosConsumerDurable(t *testing.T) {
+
+	const numMessages = 30_000
+	const timeout = 30 * time.Second // No (new) messages for 60s => terminate
+	const clusterSize = 3
+	const replicas = 3
+	const maxRetries = 25
+	const retryDelay = 500 * time.Millisecond
+	const durableConsumerName = "durable"
+
+	c := createJetStreamClusterExplicit(t, chaosConsumerTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createStreamForConsumerChaosTest(t, c, replicas, numMessages)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: 1,
+			maxDownServers: clusterSize,
+			pause:          3 * time.Second,
+		},
+	)
+
+	var nc *nats.Conn
+	var sub *nats.Subscription
+	var subLock sync.Mutex
+
+	var handleMsgFun func(msg *nats.Msg)
+	var natsURL string
+
+	{
+		var sb strings.Builder
+		for _, s := range c.servers {
+			sb.WriteString(s.ClientURL())
+			sb.WriteString(",")
+		}
+		natsURL = sb.String()
+	}
+
+	resetDurableConsumer := func() {
+		subLock.Lock()
+		defer subLock.Unlock()
+
+		if nc != nil {
+			nc.Close()
+		}
+
+		var newNc *nats.Conn
+	connectRetryLoop:
+		for r := 0; r <= maxRetries; r++ {
+			var connErr error
+			newNc, connErr = nats.Connect(natsURL)
+			if connErr == nil {
+				break connectRetryLoop
+			} else if r == maxRetries {
+				t.Fatalf("Failed to connect, exceeded max retries, last error: %s", connErr)
+			} else {
+				time.Sleep(retryDelay)
+			}
+		}
+
+		var newJs nats.JetStreamContext
+	jsRetryLoop:
+		for r := 0; r <= maxRetries; r++ {
+			var jsErr error
+			newJs, jsErr = newNc.JetStream(nats.MaxWait(10 * time.Second))
+			if jsErr == nil {
+				break jsRetryLoop
+			} else if r == maxRetries {
+				t.Fatalf("Failed to get JS, exceeded max retries, last error: %s", jsErr)
+			} else {
+				time.Sleep(retryDelay)
+			}
+		}
+
+		subOpts := []nats.SubOpt{
+			nats.Durable(durableConsumerName),
+		}
+
+		var newSub *nats.Subscription
+	subscribeRetryLoop:
+		for i := 0; i <= maxRetries; i++ {
+			var subErr error
+			newSub, subErr = newJs.Subscribe(chaosConsumerTestsSubject, handleMsgFun, subOpts...)
+			if subErr == nil {
+				ci, err := newJs.ConsumerInfo(chaosConsumerTestsStreamName, durableConsumerName)
+				if err == nil {
+					if chaosConsumerTestsDebug {
+						t.Logf("Consumer info:\n %s", toIndentedJsonString(ci))
+					}
+				} else {
+					t.Logf("Failed to retrieve consumer info: %s", err)
+				}
+
+				break subscribeRetryLoop
+			} else if i == maxRetries {
+				t.Fatalf("Exceeded max retries creating subscription: %v", subErr)
+			} else {
+				time.Sleep(retryDelay)
+			}
+		}
+
+		nc, sub = newNc, newSub
+	}
+
+	timeoutTimer := time.NewTimer(timeout)
+	deliveryCount := uint64(0)
+	received := NewBitset(numMessages)
+
+	handleMsgFun = func(msg *nats.Msg) {
+
+		subLock.Lock()
+		if msg.Sub != sub {
+			// Message from a previous instance of durable consumer, drop
+			defer subLock.Unlock()
+			return
+		}
+		subLock.Unlock()
+
+		deliveryCount += 1
+
+		metadata, err := msg.Metadata()
+		if err != nil {
+			t.Fatalf("Failed to get message metadata: %v", err)
+		}
+		seq := metadata.Sequence.Stream
+
+		var expectedMsgData = []byte(fmt.Sprintf("%d", seq))
+		if !bytes.Equal(msg.Data, expectedMsgData) {
+			t.Fatalf("Expecting message content '%s', got '%s' instead", expectedMsgData, msg.Data)
+		}
+
+		isDupe := received.get(seq - 1)
+
+		if isDupe {
+			if chaosConsumerTestsDebug {
+				t.Logf("Duplicate message delivery, seq: %d", seq)
+			}
+			return
+		}
+
+		// Mark this sequence as received
+		received.set(seq-1, true)
+		if received.count() < numMessages {
+			// Reset timeout
+			timeoutTimer.Reset(timeout)
+		} else {
+			// All received, speed up the shutdown
+			timeoutTimer.Reset(1 * time.Second)
+		}
+
+		// Simulate application processing (and gives the monkey some time to brew chaos)
+		time.Sleep(10 * time.Millisecond)
+
+	ackRetryLoop:
+		for i := 0; i <= maxRetries; i++ {
+			ackErr := msg.Ack()
+			if ackErr == nil {
+				break ackRetryLoop
+			} else if i == maxRetries {
+				t.Fatalf("Failed to ACK message %d (retried %d times)", seq, maxRetries)
+			} else {
+				time.Sleep(retryDelay)
+			}
+		}
+
+		if received.count()%1000 == 0 {
+			t.Logf("Consumed %d/%d, duplicate deliveries: %d", received.count(), numMessages, deliveryCount-received.count())
+			// Close connection and resume consuming on a different one
+			resetDurableConsumer()
+		}
+	}
+
+	resetDurableConsumer()
+
+	chaos.start()
+	defer chaos.stop()
+
+	// Wait for long enough silence.
+	// Either a stall, or all messages received
+	<-timeoutTimer.C
+
+	// Shut down consumer
+	if sub != nil {
+		sub.Unsubscribe()
+	}
+
+	uniqueDeliveredCount := received.count()
+
+	t.Logf(
+		"Delivered %d/%d messages %d duplicate deliveries",
+		uniqueDeliveredCount,
+		numMessages,
+		deliveryCount-uniqueDeliveredCount,
+	)
+
+	if uniqueDeliveredCount != numMessages {
+		t.Fatalf("No new message delivered in the last %s, %d/%d messages never delivered", timeout, numMessages-uniqueDeliveredCount, numMessages)
+	}
+}
+
+func TestJetStreamChaosConsumerPull(t *testing.T) {
+
+	const numMessages = 10_000
+	const maxRetries = 100
+	const retryDelay = 500 * time.Millisecond
+	const fetchTimeout = 250 * time.Millisecond
+	const fetchBatchSize = 100
+	const clusterSize = 3
+	const replicas = 3
+	const durableConsumerName = "durable"
+
+	c := createJetStreamClusterExplicit(t, chaosConsumerTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createStreamForConsumerChaosTest(t, c, replicas, numMessages)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: clusterSize, // Whole cluster outage
+			maxDownServers: clusterSize,
+			pause:          1 * time.Second,
+		},
+	)
+
+	subNc, subJs := jsClientConnectCluster(t, c)
+	defer subNc.Close()
+
+	sub, err := subJs.PullSubscribe(
+		chaosConsumerTestsSubject,
+		durableConsumerName,
+	)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	if chaosConsumerTestsDebug {
+		t.Logf("Initial subscription: %s", toIndentedJsonString(sub))
+	}
+
+	chaos.start()
+	defer chaos.stop()
+
+	fetchMaxWait := nats.MaxWait(fetchTimeout)
+	received := NewBitset(numMessages)
+	deliveredCount := uint64(0)
+
+	for received.count() < numMessages {
+
+		var msgs []*nats.Msg
+		var fetchErr error
+
+	fetchRetryLoop:
+		for r := 0; r <= maxRetries; r++ {
+			msgs, fetchErr = sub.Fetch(fetchBatchSize, fetchMaxWait)
+			if fetchErr == nil {
+				break fetchRetryLoop
+			} else if r == maxRetries {
+				t.Fatalf("Exceeded max retries for Fetch, last error: %s", fetchErr)
+			} else if fetchErr == nats.ErrBadSubscription {
+				t.Fatalf("Subscription is invalid: %s", toIndentedJsonString(sub))
+			} else {
+				// t.Logf("Fetch error: %v", fetchErr)
+				time.Sleep(retryDelay)
+			}
+		}
+
+		for _, msg := range msgs {
+
+			deliveredCount += 1
+
+			metadata, err := msg.Metadata()
+			if err != nil {
+				t.Fatalf("Failed to get message metadata: %v", err)
+			}
+
+			streamSeq := metadata.Sequence.Stream
+
+			expectedMsgData := []byte(fmt.Sprintf("%d", streamSeq))
+
+			if !bytes.Equal(msg.Data, expectedMsgData) {
+				t.Fatalf("Expecting message %s, got %s instead", expectedMsgData, msg.Data)
+			}
+
+			isDupe := received.get(streamSeq - 1)
+
+			received.set(streamSeq-1, true)
+
+			// Simulate application processing (and gives the monkey some time to brew chaos)
+			time.Sleep(10 * time.Millisecond)
+
+		ackRetryLoop:
+			for r := 0; r <= maxRetries; r++ {
+				ackErr := msg.Ack()
+				if ackErr == nil {
+					break ackRetryLoop
+				} else if r == maxRetries {
+					t.Fatalf("Failed to ACK message %d, last error: %s", streamSeq, ackErr)
+				} else {
+					time.Sleep(retryDelay)
+				}
+			}
+
+			if !isDupe && received.count()%1000 == 0 {
+				t.Logf("Consumed %d/%d (duplicates: %d)", received.count(), numMessages, deliveredCount-received.count())
+			}
+		}
+	}
+}
diff --git a/server/jetstream_chaos_helpers_test.go b/server/jetstream_chaos_helpers_test.go
index 5b3f9001..4c89adea 100644
--- a/server/jetstream_chaos_helpers_test.go
+++ b/server/jetstream_chaos_helpers_test.go
@@ -17,8 +17,10 @@
 package server
 
 import (
+	"encoding/json"
 	"fmt"
 	"math/rand"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -175,3 +177,90 @@ func (m *clusterBouncerChaosMonkey) run(t *testing.T, c *cluster, stopCh <-chan
 		c.waitOnLeader()
 	}
 }
+
+// Bitset, aka bitvector, allows tracking of large number of bits efficiently
+type bitset struct {
+	// Bit map storage
+	bitmap []uint8
+	// Number of bits currently set to 1
+	currentCount uint64
+	// Number of bits stored
+	size uint64
+}
+
+func NewBitset(size uint64) *bitset {
+	byteSize := (size + 7) / 8 //Round up to the nearest byte
+
+	return &bitset{
+		bitmap:       make([]uint8, int(byteSize)),
+		size:         size,
+		currentCount: 0,
+	}
+}
+
+func (b *bitset) get(index uint64) bool {
+	if index >= b.size {
+		panic(fmt.Sprintf("Index %d out of bounds, size %d", index, b.size))
+	}
+	byteIndex := index / 8
+	bitIndex := uint(index % 8)
+	bit := (b.bitmap[byteIndex] & (uint8(1) << bitIndex))
+	return bit != 0
+}
+
+func (b *bitset) set(index uint64, value bool) {
+	if index >= b.size {
+		panic(fmt.Sprintf("Index %d out of bounds, size %d", index, b.size))
+	}
+	byteIndex := index / 8
+	bitIndex := uint(index % 8)
+	byteMask := uint8(1) << bitIndex
+	isSet := (b.bitmap[byteIndex] & (uint8(1) << bitIndex)) != 0
+	if value {
+		b.bitmap[byteIndex] |= byteMask
+		if !isSet {
+			b.currentCount += 1
+		}
+	} else {
+		b.bitmap[byteIndex] &= ^byteMask
+		if isSet {
+			b.currentCount -= 1
+		}
+	}
+}
+
+func (b *bitset) count() uint64 {
+	return b.currentCount
+}
+
+func (b *bitset) String() string {
+	const block = 8 // 8 bytes, 64 bits per line
+	sb := strings.Builder{}
+
+	sb.WriteString(fmt.Sprintf("Bits set: %d/%d\n", b.currentCount, b.size))
+	for i := 0; i < len(b.bitmap); i++ {
+		if i%block == 0 {
+			if i > 0 {
+				sb.WriteString("\n")
+			}
+			sb.WriteString(fmt.Sprintf("[%4d] ", i*8))
+		}
+		for j := uint8(0); j < 8; j++ {
+			if b.bitmap[i]&(1<<j) > 0 {
+				sb.WriteString("1")
+			} else {
+				sb.WriteString("0")
+			}
+		}
+	}
+	sb.WriteString("\n")
+	return sb.String()
+}
+
+func toIndentedJsonString(v interface{}) string {
+	jsonBytes, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		return fmt.Sprintf("Marshal error: %s", err)
+	}
+	return string(jsonBytes)
+}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 1368dc19..a45edee4 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -5431,9 +5431,22 @@ func TestJetStreamClusterSourcesFilterSubjectUpdate(t *testing.T) {
 		Replicas: 2,
 	})
 	require_NoError(t, err)
-	// no send was necessary as we received previously filtered messages
-	checkSync(500, 300)
+	// The filter was completely switched, which is why we only receive new messages
+	checkSync(500, 200)
+	sendBatch("foo", 100)
+	checkSync(600, 300)
+	sendBatch("bar", 100)
+	checkSync(700, 300)
 
+	// change filter subject to *, as the internal sequence number does not cover the previously filtered tail end
+	_, err = js.UpdateStream(&nats.StreamConfig{
+		Name:     "M",
+		Sources:  []*nats.StreamSource{{Name: "TEST", FilterSubject: "*"}},
+		Replicas: 2,
+	})
+	require_NoError(t, err)
+	// no send was necessary as we received previously filtered messages
+	checkSync(700, 400)
 }
 
 func TestJetStreamClusterSourcesUpdateOriginError(t *testing.T) {
@@ -7229,7 +7242,7 @@ func TestJetStreamClusterLeafDifferentAccounts(t *testing.T) {
 	nc, js := jsClientConnect(t, ln.randomServer())
 	defer nc.Close()
 
-	// Make sure we can properly indentify the right account when the leader received the request.
+	// Make sure we can properly identify the right account when the leader received the request.
 	// We need to map the client info header to the new account once received by the hub.
 	if _, err := js.AccountInfo(); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -11672,6 +11685,65 @@ func TestJetStreamClusterConsumerDeliverNewNotConsumingBeforeStepDownOrRestart(t
 	checkCount(0)
 }
 
+func TestJetStreamClusterConsumerDeliverNewMaxRedeliveriesAndServerRestart(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",
+		Subjects: []string{"foo.*"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	inbox := nats.NewInbox()
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		DeliverSubject: inbox,
+		Durable:        "dur",
+		AckPolicy:      nats.AckExplicitPolicy,
+		DeliverPolicy:  nats.DeliverNewPolicy,
+		MaxDeliver:     3,
+		AckWait:        250 * time.Millisecond,
+		FilterSubject:  "foo.bar",
+	})
+	require_NoError(t, err)
+
+	c.waitOnConsumerLeader(globalAccountName, "TEST", "dur")
+
+	sendStreamMsg(t, nc, "foo.bar", "msg")
+
+	sub := natsSubSync(t, nc, inbox)
+	for i := 0; i < 3; i++ {
+		natsNexMsg(t, sub, time.Second)
+	}
+	// Now check that there is no more redeliveries
+	if msg, err := sub.NextMsg(300 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Expected timeout, got msg=%+v err=%v", msg, err)
+	}
+
+	// Give a chance to things to be persisted
+	time.Sleep(300 * time.Millisecond)
+
+	// Check server restart
+	nc.Close()
+	c.stopAll()
+	c.restartAll()
+
+	c.waitOnConsumerLeader(globalAccountName, "TEST", "dur")
+
+	nc, _ = jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	sub = natsSubSync(t, nc, inbox)
+	// We should not have messages being redelivered.
+	if msg, err := sub.NextMsg(300 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Expected timeout, got msg=%+v err=%v", msg, err)
+	}
+}
+
 func TestJetStreamClusterNoRestartAdvisories(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index 96f8691f..954cdbb5 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -1068,6 +1068,33 @@ func jsClientConnectEx(t *testing.T, s *Server, domain string, opts ...nats.Opti
 	return nc, js
 }
 
+func jsClientConnectCluster(t *testing.T, c *cluster, opts ...nats.Option) (*nats.Conn, nats.JetStreamContext) {
+	t.Helper()
+
+	var sb strings.Builder
+
+	for _, s := range c.servers {
+		sb.WriteString(s.ClientURL())
+		sb.WriteString(",")
+	}
+
+	return jsClientConnectURL(t, sb.String(), opts...)
+}
+
+func jsClientConnectURL(t *testing.T, url string, opts ...nats.Option) (*nats.Conn, nats.JetStreamContext) {
+	t.Helper()
+
+	nc, err := nats.Connect(url, opts...)
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error getting JetStream context: %v", err)
+	}
+	return nc, js
+}
+
 func checkSubsPending(t *testing.T, sub *nats.Subscription, numExpected int) {
 	t.Helper()
 	checkFor(t, 10*time.Second, 20*time.Millisecond, func() error {
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index 8d7a1ad4..89ddc218 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -1778,7 +1778,7 @@ func TestJetStreamSuperClusterMovingStreamsAndConsumers(t *testing.T) {
 				Replicas:  replicas,
 				Placement: &nats.Placement{Tags: []string{"cloud:aws"}},
 			})
-			require_Error(t, err, NewJSStreamMoveInProgressError())
+			require_Contains(t, err.Error(), "stream move already in progress")
 
 			checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
 				si, err := js.StreamInfo("MOVE")
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 6c080c63..8974a08c 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -12338,203 +12338,218 @@ func TestJetStreamTemplatedErrorsBug(t *testing.T) {
 }
 
 func TestJetStreamServerEncryption(t *testing.T) {
-	tmpl := `
-		listen: 127.0.0.1:-1
-		jetstream: {key: $JS_KEY, store_dir: '%s'}
-	`
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer removeDir(t, storeDir)
+	cases := []struct {
+		name   string
+		cstr   string
+		cipher StoreCipher
+	}{
+		{"Default", _EMPTY_, ChaCha},
+		{"ChaCha", ", cipher: chacha", ChaCha},
+		{"AES", ", cipher: aes", AES},
+	}
 
-	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, storeDir)))
-	defer removeFile(t, conf)
+	for _, c := range cases {
 
-	os.Setenv("JS_KEY", "s3cr3t!!")
-	defer os.Unsetenv("JS_KEY")
+		t.Run(c.name, func(t *testing.T) {
+			tmpl := `
+				listen: 127.0.0.1:-1
+				jetstream: {key: $JS_KEY, store_dir: '%s' %s}
+			`
+			storeDir := createDir(t, JetStreamStoreDir)
+			defer removeDir(t, storeDir)
 
-	s, _ := RunServerWithConfig(conf)
-	defer s.Shutdown()
+			conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, storeDir, c.cstr)))
+			defer removeFile(t, conf)
 
-	config := s.JetStreamConfig()
-	if config == nil {
-		t.Fatalf("Expected config but got none")
-	}
-	defer removeDir(t, config.StoreDir)
+			os.Setenv("JS_KEY", "s3cr3t!!")
+			defer os.Unsetenv("JS_KEY")
 
-	// Client based API
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
+			s, _ := RunServerWithConfig(conf)
+			defer s.Shutdown()
 
-	cfg := &nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"foo", "bar", "baz"},
-	}
-	if _, err := js.AddStream(cfg); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+			config := s.JetStreamConfig()
+			if config == nil {
+				t.Fatalf("Expected config but got none")
+			}
+			defer removeDir(t, config.StoreDir)
 
-	msg := []byte("ENCRYPTED PAYLOAD!!")
-	sendMsg := func(subj string) {
-		t.Helper()
-		if _, err := js.Publish(subj, msg); err != nil {
-			t.Fatalf("Unexpected publish error: %v", err)
-		}
-	}
-	// Send 10 msgs
-	for i := 0; i < 10; i++ {
-		sendMsg("foo")
-	}
+			// Client based API
+			nc, js := jsClientConnect(t, s)
+			defer nc.Close()
 
-	// Now create a consumer.
-	sub, err := js.PullSubscribe("foo", "dlc")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	for i, m := range fetchMsgs(t, sub, 10, 5*time.Second) {
-		if i < 5 {
-			m.AckSync()
-		}
-	}
+			cfg := &nats.StreamConfig{
+				Name:     "TEST",
+				Subjects: []string{"foo", "bar", "baz"},
+			}
+			if _, err := js.AddStream(cfg); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 
-	// Grab our state to compare after restart.
-	si, _ := js.StreamInfo("TEST")
-	ci, _ := js.ConsumerInfo("TEST", "dlc")
+			msg := []byte("ENCRYPTED PAYLOAD!!")
+			sendMsg := func(subj string) {
+				t.Helper()
+				if _, err := js.Publish(subj, msg); err != nil {
+					t.Fatalf("Unexpected publish error: %v", err)
+				}
+			}
+			// Send 10 msgs
+			for i := 0; i < 10; i++ {
+				sendMsg("foo")
+			}
 
-	// Quick check to make sure everything not just plaintext still.
-	sdir := filepath.Join(config.StoreDir, "$G", "streams", "TEST")
-	// Make sure we can not find any plaintext strings in the target file.
-	checkFor := func(fn string, strs ...string) {
-		t.Helper()
-		data, err := os.ReadFile(fn)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		for _, str := range strs {
-			if bytes.Contains(data, []byte(str)) {
-				t.Fatalf("Found %q in body of file contents", str)
+			// Now create a consumer.
+			sub, err := js.PullSubscribe("foo", "dlc")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			for i, m := range fetchMsgs(t, sub, 10, 5*time.Second) {
+				if i < 5 {
+					m.AckSync()
+				}
 			}
-		}
-	}
-	checkKeyFile := func(fn string) {
-		t.Helper()
-		if _, err := os.Stat(fn); err != nil {
-			t.Fatalf("Expected a key file at %q", fn)
-		}
-	}
 
-	// Check stream meta.
-	checkEncrypted := func() {
-		checkKeyFile(filepath.Join(sdir, JetStreamMetaFileKey))
-		checkFor(filepath.Join(sdir, JetStreamMetaFile), "TEST", "foo", "bar", "baz", "max_msgs", "max_bytes")
-		// Check a message block.
-		checkKeyFile(filepath.Join(sdir, "msgs", "1.key"))
-		checkFor(filepath.Join(sdir, "msgs", "1.blk"), "ENCRYPTED PAYLOAD!!", "foo", "bar", "baz")
+			// Grab our state to compare after restart.
+			si, _ := js.StreamInfo("TEST")
+			ci, _ := js.ConsumerInfo("TEST", "dlc")
 
-		// Check consumer meta and state.
-		checkKeyFile(filepath.Join(sdir, "obs", "dlc", JetStreamMetaFileKey))
-		checkFor(filepath.Join(sdir, "obs", "dlc", JetStreamMetaFile), "TEST", "dlc", "foo", "bar", "baz", "max_msgs", "ack_policy")
-		// Load and see if we can parse the consumer state.
-		state, err := os.ReadFile(filepath.Join(sdir, "obs", "dlc", "o.dat"))
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if _, err := decodeConsumerState(state); err == nil {
-			t.Fatalf("Expected decoding consumer state to fail")
-		}
-	}
+			// Quick check to make sure everything not just plaintext still.
+			sdir := filepath.Join(config.StoreDir, "$G", "streams", "TEST")
+			// Make sure we can not find any plaintext strings in the target file.
+			checkFor := func(fn string, strs ...string) {
+				t.Helper()
+				data, err := os.ReadFile(fn)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				for _, str := range strs {
+					if bytes.Contains(data, []byte(str)) {
+						t.Fatalf("Found %q in body of file contents", str)
+					}
+				}
+			}
+			checkKeyFile := func(fn string) {
+				t.Helper()
+				if _, err := os.Stat(fn); err != nil {
+					t.Fatalf("Expected a key file at %q", fn)
+				}
+			}
 
-	// Stop current
-	s.Shutdown()
+			// Check stream meta.
+			checkEncrypted := func() {
+				checkKeyFile(filepath.Join(sdir, JetStreamMetaFileKey))
+				checkFor(filepath.Join(sdir, JetStreamMetaFile), "TEST", "foo", "bar", "baz", "max_msgs", "max_bytes")
+				// Check a message block.
+				checkKeyFile(filepath.Join(sdir, "msgs", "1.key"))
+				checkFor(filepath.Join(sdir, "msgs", "1.blk"), "ENCRYPTED PAYLOAD!!", "foo", "bar", "baz")
 
-	checkEncrypted()
+				// Check consumer meta and state.
+				checkKeyFile(filepath.Join(sdir, "obs", "dlc", JetStreamMetaFileKey))
+				checkFor(filepath.Join(sdir, "obs", "dlc", JetStreamMetaFile), "TEST", "dlc", "foo", "bar", "baz", "max_msgs", "ack_policy")
+				// Load and see if we can parse the consumer state.
+				state, err := os.ReadFile(filepath.Join(sdir, "obs", "dlc", "o.dat"))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				if _, err := decodeConsumerState(state); err == nil {
+					t.Fatalf("Expected decoding consumer state to fail")
+				}
+			}
 
-	// Restart.
-	s, _ = RunServerWithConfig(conf)
-	defer s.Shutdown()
+			// Stop current
+			s.Shutdown()
 
-	// Connect again.
-	nc, js = jsClientConnect(t, s)
-	defer nc.Close()
+			checkEncrypted()
 
-	si2, err := js.StreamInfo("TEST")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	if !reflect.DeepEqual(si, si2) {
-		t.Fatalf("Stream infos did not match\n%+v\nvs\n%+v", si, si2)
-	}
+			// Restart.
+			s, _ = RunServerWithConfig(conf)
+			defer s.Shutdown()
 
-	ci2, _ := js.ConsumerInfo("TEST", "dlc")
-	// Consumer create times can be slightly off after restore from disk.
-	now := time.Now()
-	ci.Created, ci2.Created = now, now
-	ci.Delivered.Last, ci2.Delivered.Last = nil, nil
-	ci.AckFloor.Last, ci2.AckFloor.Last = nil, nil
-	// Also clusters will be different.
-	ci.Cluster, ci2.Cluster = nil, nil
-	if !reflect.DeepEqual(ci, ci2) {
-		t.Fatalf("Consumer infos did not match\n%+v\nvs\n%+v", ci, ci2)
-	}
+			// Connect again.
+			nc, js = jsClientConnect(t, s)
+			defer nc.Close()
 
-	// Send 10 more msgs
-	for i := 0; i < 10; i++ {
-		sendMsg("foo")
-	}
-	if si, err = js.StreamInfo("TEST"); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	if si.State.Msgs != 20 {
-		t.Fatalf("Expected 20 msgs total, got %d", si.State.Msgs)
-	}
+			si2, err := js.StreamInfo("TEST")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if !reflect.DeepEqual(si, si2) {
+				t.Fatalf("Stream infos did not match\n%+v\nvs\n%+v", si, si2)
+			}
 
-	// Now test snapshots etc.
-	acc := s.GlobalAccount()
-	mset, err := acc.lookupStream("TEST")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	scfg := mset.config()
-	sr, err := mset.snapshot(5*time.Second, false, true)
-	if err != nil {
-		t.Fatalf("Error getting snapshot: %v", err)
-	}
-	snapshot, err := io.ReadAll(sr.Reader)
-	if err != nil {
-		t.Fatalf("Error reading snapshot")
-	}
+			ci2, _ := js.ConsumerInfo("TEST", "dlc")
+			// Consumer create times can be slightly off after restore from disk.
+			now := time.Now()
+			ci.Created, ci2.Created = now, now
+			ci.Delivered.Last, ci2.Delivered.Last = nil, nil
+			ci.AckFloor.Last, ci2.AckFloor.Last = nil, nil
+			// Also clusters will be different.
+			ci.Cluster, ci2.Cluster = nil, nil
+			if !reflect.DeepEqual(ci, ci2) {
+				t.Fatalf("Consumer infos did not match\n%+v\nvs\n%+v", ci, ci2)
+			}
 
-	// Run new server w/o encryption. Make sure we can restore properly (meaning encryption was stripped etc).
-	ns := RunBasicJetStreamServer()
-	defer ns.Shutdown()
+			// Send 10 more msgs
+			for i := 0; i < 10; i++ {
+				sendMsg("foo")
+			}
+			if si, err = js.StreamInfo("TEST"); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if si.State.Msgs != 20 {
+				t.Fatalf("Expected 20 msgs total, got %d", si.State.Msgs)
+			}
 
-	nacc := ns.GlobalAccount()
-	r := bytes.NewReader(snapshot)
-	mset, err = nacc.RestoreStream(&scfg, r)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	ss := mset.store.State()
-	if ss.Msgs != si.State.Msgs || ss.FirstSeq != si.State.FirstSeq || ss.LastSeq != si.State.LastSeq {
-		t.Fatalf("Stream states do not match: %+v vs %+v", ss, si.State)
-	}
+			// Now test snapshots etc.
+			acc := s.GlobalAccount()
+			mset, err := acc.lookupStream("TEST")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			scfg := mset.config()
+			sr, err := mset.snapshot(5*time.Second, false, true)
+			if err != nil {
+				t.Fatalf("Error getting snapshot: %v", err)
+			}
+			snapshot, err := io.ReadAll(sr.Reader)
+			if err != nil {
+				t.Fatalf("Error reading snapshot")
+			}
 
-	// Now restore to our encrypted server as well.
-	if err := js.DeleteStream("TEST"); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+			// Run new server w/o encryption. Make sure we can restore properly (meaning encryption was stripped etc).
+			ns := RunBasicJetStreamServer()
+			defer ns.Shutdown()
 
-	acc = s.GlobalAccount()
-	r.Reset(snapshot)
-	mset, err = acc.RestoreStream(&scfg, r)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	ss = mset.store.State()
-	if ss.Msgs != si.State.Msgs || ss.FirstSeq != si.State.FirstSeq || ss.LastSeq != si.State.LastSeq {
-		t.Fatalf("Stream states do not match: %+v vs %+v", ss, si.State)
-	}
+			nacc := ns.GlobalAccount()
+			r := bytes.NewReader(snapshot)
+			mset, err = nacc.RestoreStream(&scfg, r)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			ss := mset.store.State()
+			if ss.Msgs != si.State.Msgs || ss.FirstSeq != si.State.FirstSeq || ss.LastSeq != si.State.LastSeq {
+				t.Fatalf("Stream states do not match: %+v vs %+v", ss, si.State)
+			}
 
-	// Check that all is encrypted like above since we know we need to convert since snapshots always plaintext.
-	checkEncrypted()
+			// Now restore to our encrypted server as well.
+			if err := js.DeleteStream("TEST"); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			acc = s.GlobalAccount()
+			r.Reset(snapshot)
+			mset, err = acc.RestoreStream(&scfg, r)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			ss = mset.store.State()
+			if ss.Msgs != si.State.Msgs || ss.FirstSeq != si.State.FirstSeq || ss.LastSeq != si.State.LastSeq {
+				t.Fatalf("Stream states do not match: %+v vs %+v", ss, si.State)
+			}
+
+			// Check that all is encrypted like above since we know we need to convert since snapshots always plaintext.
+			checkEncrypted()
+		})
+	}
 }
 
 // User report of bug.
@@ -18908,3 +18923,185 @@ func TestJetStreamDirectGetAutoSet(t *testing.T) {
 		t.Fatalf("Expected to see the mirror respond at least once")
 	}
 }
+
+func TestJetStreamProperErrorDueToOverlapSubjects(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	test := func(t *testing.T, s *Server) {
+		nc, js := jsClientConnect(t, s)
+		defer nc.Close()
+
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     "TEST",
+			Subjects: []string{"foo.*"},
+		})
+		require_NoError(t, err)
+
+		// Now do this by end since we want to check the error returned.
+		sc := &nats.StreamConfig{
+			Name:     "TEST2",
+			Subjects: []string{"foo.>"},
+		}
+		req, _ := json.Marshal(sc)
+		msg, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, sc.Name), req, time.Second)
+		require_NoError(t, err)
+
+		var scResp JSApiStreamCreateResponse
+		err = json.Unmarshal(msg.Data, &scResp)
+		require_NoError(t, err)
+
+		if scResp.Error == nil || !IsNatsErr(scResp.Error, JSStreamSubjectOverlapErr) {
+			t.Fatalf("Did not receive correct error: %+v", scResp)
+		}
+	}
+
+	t.Run("standalone", func(t *testing.T) { test(t, s) })
+	t.Run("clustered", func(t *testing.T) { test(t, c.randomServer()) })
+}
+
+func TestJetStreamServerCipherConvert(t *testing.T) {
+	tmpl := `
+		listen: 127.0.0.1:-1
+		jetstream: {key: s3cr3t, store_dir: '%s', cipher: %s}
+	`
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	// Create a stream and a consumer under one cipher, and restart the server with a new cipher.
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, storeDir, "AES")))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// Client based API
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	}
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	for i := 0; i < 1000; i++ {
+		msg := []byte(fmt.Sprintf("TOP SECRET DOCUMENT #%d", i+1))
+		_, err := js.Publish("foo", msg)
+		require_NoError(t, err)
+	}
+
+	// Make sure consumers convert as well.
+	sub, err := js.PullSubscribe("foo", "dlc")
+	require_NoError(t, err)
+	for _, m := range fetchMsgs(t, sub, 100, 5*time.Second) {
+		m.AckSync()
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+
+	ci, err := js.ConsumerInfo("TEST", "dlc")
+	require_NoError(t, err)
+
+	// Stop current
+	s.Shutdown()
+
+	conf = createConfFile(t, []byte(fmt.Sprintf(tmpl, storeDir, "ChaCha")))
+	defer removeFile(t, conf)
+
+	s, _ = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, js = jsClientConnect(t, s)
+	defer nc.Close()
+
+	si2, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+
+	if !reflect.DeepEqual(si, si2) {
+		t.Fatalf("Stream infos did not match\n%+v\nvs\n%+v", si, si2)
+	}
+
+	ci2, err := js.ConsumerInfo("TEST", "dlc")
+	require_NoError(t, err)
+
+	// Consumer create times can be slightly off after restore from disk.
+	now := time.Now()
+	ci.Created, ci2.Created = now, now
+	ci.Delivered.Last, ci2.Delivered.Last = nil, nil
+	ci.AckFloor.Last, ci2.AckFloor.Last = nil, nil
+	// Also clusters will be different.
+	ci.Cluster, ci2.Cluster = nil, nil
+
+	if !reflect.DeepEqual(ci, ci2) {
+		t.Fatalf("Consumer infos did not match\n%+v\nvs\n%+v", ci, ci2)
+	}
+}
+
+func TestJetStreamConsumerDeliverNewMaxRedeliveriesAndServerRestart(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo.*"},
+	})
+	require_NoError(t, err)
+
+	inbox := nats.NewInbox()
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		DeliverSubject: inbox,
+		Durable:        "dur",
+		AckPolicy:      nats.AckExplicitPolicy,
+		DeliverPolicy:  nats.DeliverNewPolicy,
+		MaxDeliver:     3,
+		AckWait:        250 * time.Millisecond,
+		FilterSubject:  "foo.bar",
+	})
+	require_NoError(t, err)
+
+	sendStreamMsg(t, nc, "foo.bar", "msg")
+
+	sub := natsSubSync(t, nc, inbox)
+	for i := 0; i < 3; i++ {
+		natsNexMsg(t, sub, time.Second)
+	}
+	// Now check that there is no more redeliveries
+	if msg, err := sub.NextMsg(300 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Expected timeout, got msg=%+v err=%v", msg, err)
+	}
+
+	// Give a chance to things to be persisted
+	time.Sleep(300 * time.Millisecond)
+
+	// Check server restart
+	nc.Close()
+	sd := s.JetStreamConfig().StoreDir
+	s.Shutdown()
+	s = RunJetStreamServerOnPort(-1, sd)
+	defer s.Shutdown()
+
+	nc, _ = jsClientConnect(t, s)
+	defer nc.Close()
+
+	sub = natsSubSync(t, nc, inbox)
+	// We should not have messages being redelivered.
+	if msg, err := sub.NextMsg(300 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Expected timeout, got msg=%+v err=%v", msg, err)
+	}
+}
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 561b7e95..5b498445 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -4163,6 +4163,124 @@ func TestJWTLimits(t *testing.T) {
 	})
 }
 
+func TestJwtTemplates(t *testing.T) {
+	kp, _ := nkeys.CreateAccount()
+	aPub, _ := kp.PublicKey()
+	ukp, _ := nkeys.CreateUser()
+	upub, _ := ukp.PublicKey()
+	uclaim := newJWTTestUserClaims()
+	uclaim.Name = "myname"
+	uclaim.Subject = upub
+	uclaim.SetScoped(true)
+	uclaim.IssuerAccount = aPub
+	uclaim.Tags.Add("foo:foo1")
+	uclaim.Tags.Add("foo:foo2")
+	uclaim.Tags.Add("bar:bar1")
+	uclaim.Tags.Add("bar:bar2")
+	uclaim.Tags.Add("bar:bar3")
+
+	lim := jwt.UserPermissionLimits{}
+	lim.Pub.Allow.Add("{{tag(foo)}}.none.{{tag(bar)}}")
+	lim.Pub.Deny.Add("{{tag(foo)}}.{{account-tag(acc)}}")
+	lim.Sub.Allow.Add("{{tag(NOT_THERE)}}") // expect to not emit this
+	lim.Sub.Deny.Add("foo.{{name()}}.{{subject()}}.{{account-name()}}.{{account-subject()}}.bar")
+	acc := &Account{nameTag: "accname", tags: []string{"acc:acc1", "acc:acc2"}}
+
+	resLim, err := processUserPermissionsTemplate(lim, uclaim, acc)
+	require_NoError(t, err)
+
+	test := func(expectedSubjects []string, res jwt.StringList) {
+		t.Helper()
+		require_True(t, len(res) == len(expectedSubjects))
+		for _, expetedSubj := range expectedSubjects {
+			require_True(t, res.Contains(expetedSubj))
+		}
+	}
+
+	test(resLim.Pub.Allow, []string{"foo1.none.bar1", "foo1.none.bar2", "foo1.none.bar3",
+		"foo2.none.bar1", "foo2.none.bar2", "foo2.none.bar3"})
+
+	test(resLim.Pub.Deny, []string{"foo1.acc1", "foo1.acc2", "foo2.acc1", "foo2.acc2"})
+
+	require_True(t, len(resLim.Sub.Allow) == 0)
+	require_True(t, len(resLim.Sub.Deny) == 1)
+	require_Contains(t, resLim.Sub.Deny[0], fmt.Sprintf("foo.myname.%s.accname.%s.bar", upub, aPub))
+
+	lim.Pub.Deny.Add("{{tag(NOT_THERE)}}")
+	_, err = processUserPermissionsTemplate(lim, uclaim, acc)
+	require_Error(t, err)
+	require_Contains(t, err.Error(), "generated invalid subject")
+}
+
+func TestJWTLimitsTemplate(t *testing.T) {
+	kp, _ := nkeys.CreateAccount()
+	aPub, _ := kp.PublicKey()
+	claim := jwt.NewAccountClaims(aPub)
+	aSignScopedKp, aSignScopedPub := createKey(t)
+	signer := jwt.NewUserScope()
+	signer.Key = aSignScopedPub
+	signer.Template.Pub.Deny.Add("denied")
+	signer.Template.Pub.Allow.Add("foo.{{name()}}")
+	signer.Template.Sub.Allow.Add("foo.{{name()}}")
+	claim.SigningKeys.AddScopedSigner(signer)
+	aJwt, err := claim.Encode(oKp)
+	require_NoError(t, err)
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		operator: %s
+		resolver: MEM
+		resolver_preload: {
+			%s: %s
+		}
+    `, ojwt, aPub, aJwt)))
+	defer removeFile(t, conf)
+	sA, _ := RunServerWithConfig(conf)
+	defer sA.Shutdown()
+	errChan := make(chan struct{})
+	defer close(errChan)
+
+	ukp, _ := nkeys.CreateUser()
+	seed, _ := ukp.Seed()
+	upub, _ := ukp.PublicKey()
+	uclaim := newJWTTestUserClaims()
+	uclaim.Name = "myname"
+	uclaim.Subject = upub
+	uclaim.SetScoped(true)
+	uclaim.IssuerAccount = aPub
+
+	ujwt, err := uclaim.Encode(aSignScopedKp)
+	require_NoError(t, err)
+	creds := genCredsFile(t, ujwt, seed)
+
+	defer removeFile(t, creds)
+
+	t.Run("pass", func(t *testing.T) {
+		c := natsConnect(t, sA.ClientURL(), nats.UserCredentials(creds))
+		defer c.Close()
+		sub, err := c.SubscribeSync("foo.myname")
+		require_NoError(t, err)
+		require_NoError(t, c.Flush())
+		require_NoError(t, c.Publish("foo.myname", nil))
+		_, err = sub.NextMsg(time.Second)
+		require_NoError(t, err)
+	})
+	t.Run("fail", func(t *testing.T) {
+		c := natsConnect(t, sA.ClientURL(), nats.UserCredentials(creds),
+			nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+				if strings.Contains(err.Error(), `nats: Permissions Violation for Publish to "foo.othername"`) {
+					errChan <- struct{}{}
+				}
+			}))
+		defer c.Close()
+		require_NoError(t, c.Publish("foo.othername", nil))
+		select {
+		case <-errChan:
+		case <-time.After(time.Second * 2):
+			require_True(t, false)
+		}
+	})
+}
+
 func TestJWTNoOperatorMode(t *testing.T) {
 	for _, login := range []bool{true, false} {
 		t.Run("", func(t *testing.T) {
diff --git a/server/norace_test.go b/server/norace_test.go
index 4dc08125..c0ba0347 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -2183,7 +2183,7 @@ func TestNoRaceJetStreamClusterMirrorExpirationAndMissingSequences(t *testing.T)
 	checkMirror(20)
 }
 
-func TestNoRaceLargeActiveOnReplica(t *testing.T) {
+func TestNoRaceJetStreamClusterLargeActiveOnReplica(t *testing.T) {
 	// Uncomment to run.
 	skip(t)
 
@@ -3562,6 +3562,8 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 	}
 	// Grab underlying raft node and the WAL (filestore) and we will attempt to "corrupt" it.
 	node := o.raftNode().(*raft)
+	// We are doing a stop here to prevent the internal consumer snapshot from happening on exit
+	node.Stop()
 	fs := node.wal.(*fileStore)
 	fcfg, cfg := fs.fcfg, fs.cfg.StreamConfig
 	// Stop all the servers.
@@ -4107,7 +4109,7 @@ func TestNoRaceJetStreamMemstoreWithLargeInteriorDeletes(t *testing.T) {
 // This is related to an issue reported where we were exhausting threads by trying to
 // cleanup too many consumers at the same time.
 // https://github.com/nats-io/nats-server/issues/2742
-func TestNoRaceConsumerFileStoreConcurrentDiskIO(t *testing.T) {
+func TestNoRaceJetStreamConsumerFileStoreConcurrentDiskIO(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
@@ -4451,7 +4453,7 @@ func TestNoRaceJetStreamConsumerFilterPerfDegradation(t *testing.T) {
 	}
 }
 
-func TestNoRaceFileStoreKeyFileCleanup(t *testing.T) {
+func TestNoRaceJetStreamFileStoreKeyFileCleanup(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
@@ -4498,7 +4500,7 @@ func TestNoRaceFileStoreKeyFileCleanup(t *testing.T) {
 	}
 }
 
-func TestNoRaceMsgIdPerfDuringCatchup(t *testing.T) {
+func TestNoRaceJetStreamMsgIdPerfDuringCatchup(t *testing.T) {
 	// Uncomment to run. Needs to be on a bigger machine. Do not want as part of Travis tests atm.
 	skip(t)
 
@@ -4582,7 +4584,7 @@ func TestNoRaceMsgIdPerfDuringCatchup(t *testing.T) {
 	}
 }
 
-func TestNoRaceRebuildDeDupeAndMemoryPerf(t *testing.T) {
+func TestNoRaceJetStreamRebuildDeDupeAndMemoryPerf(t *testing.T) {
 	skip(t)
 
 	s := RunBasicJetStreamServer()
@@ -4666,7 +4668,7 @@ func TestNoRaceRebuildDeDupeAndMemoryPerf(t *testing.T) {
 	fmt.Printf("Memory: %v\n", friendlyBytes(v.Mem))
 }
 
-func TestNoRaceMemoryUsageOnLimitedStreamWithMirror(t *testing.T) {
+func TestNoRaceJetStreamMemoryUsageOnLimitedStreamWithMirror(t *testing.T) {
 	skip(t)
 
 	s := RunBasicJetStreamServer()
@@ -4719,7 +4721,7 @@ func TestNoRaceMemoryUsageOnLimitedStreamWithMirror(t *testing.T) {
 	fmt.Printf("Memory AFTER SEND: %v\n", friendlyBytes(v.Mem))
 }
 
-func TestNoRaceOrderedConsumerLongRTTPerformance(t *testing.T) {
+func TestNoRaceJetStreamOrderedConsumerLongRTTPerformance(t *testing.T) {
 	skip(t)
 
 	s := RunBasicJetStreamServer()
@@ -5384,7 +5386,7 @@ func TestNoRaceJetStreamClusterStreamNamesAndInfosMoreThanAPILimit(t *testing.T)
 	check(JSApiStreamList, JSApiListLimit)
 }
 
-func TestNoRaceFileStoreLargeKVAccessTiming(t *testing.T) {
+func TestNoRaceJetStreamFileStoreLargeKVAccessTiming(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)