Repository: nats-io/nats-server
Pull request #3305 (2022-07-29 00:37:46)

- Changes to make adding new mapping functions easier
- Adds new subject mapping functions:
```go
{{SplitFromLeft(wildcard index, position)}}
{{SplitFromRight(wildcard index, position)}}
{{SliceFromLeft(wildcard index, slice size)}}
{{SliceFromRight(wildcard index, slice size)}}
{{Split(wildcard index, deliminator)}}
```

Examples:
```go
	shouldMatch("*", "{{splitfromleft(1,3)}}", "12345", "123.45")
	shouldMatch("*", "{{SplitFromRight(1,3)}}", "12345", "12.345")
	shouldMatch("*", "{{SliceFromLeft(1,3)}}", "1234567890", "123.456.789.0")
	shouldMatch("*", "{{SliceFromRight(1,3)}}", "1234567890", "1.234.567.890")
	shouldMatch("*", "{{split(1,-)}}", "-abc-def--ghi-", "abc.def.ghi")
	shouldMatch("*.*", "{{split(2,-)}}.{{splitfromleft(1,2)}}", "foo.-abc-def--ghij-", "abc.def.ghij.fo.o")
```

- Subject mapping functions can now be all lower case or Pascal case (or a combination): e.g. splitfromleft, SplitFromLeft, splitFromleft, etc...

 - [ ] Documentation added (if applicable)
 - [X] Tests added
 - [X] Branch rebased on top of current main (`git pull --rebase origin main`)
 - [X] Changes squashed to a single commit (described [here](http://gitready.com/advanced/2009/02/10/squashing-commits-with-rebase.html))
 - [X] Build is green in Travis CI
 - [X] 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)

/cc @nats-io/core


derekcollison: Are we going to try to merge this before 2.9 is released?
ripienaar: would love to see it merged.

Patch:
diff --git a/server/accounts.go b/server/accounts.go
index a7081f25..eb4e2ca2 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -166,8 +166,26 @@ const (
 )
 
 var commaSeparatorRegEx = regexp.MustCompile(`,\s*`)
-var partitionMappingFunctionRegEx = regexp.MustCompile(`{{\s*partition\s*\((.*)\)\s*}}`)
-var wildcardMappingFunctionRegEx = regexp.MustCompile(`{{\s*wildcard\s*\((.*)\)\s*}}`)
+var partitionMappingFunctionRegEx = regexp.MustCompile(`{{\s*[pP]artition\s*\((.*)\)\s*}}`)
+var wildcardMappingFunctionRegEx = regexp.MustCompile(`{{\s*[wW]ildcard\s*\((.*)\)\s*}}`)
+var splitFromLeftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
+var splitFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
+var sliceFromLeftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
+var sliceFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
+var splitMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit\s*\((.*)\)\s*}}`)
+
+// Enum for the subject mapping transform function types
+const (
+	NoTransform int16 = iota
+	BadTransform
+	Partition
+	Wildcard
+	SplitFromLeft
+	SplitFromRight
+	SliceFromLeft
+	SliceFromRight
+	Split
+)
 
 // String helper.
 func (rt ServiceRespType) String() string {
@@ -845,7 +863,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 	}
 
 	if d != nil {
-		if len(d.tr.dtpi) == 0 {
+		if len(d.tr.dtokmftokindexesargs) == 0 {
 			ndest = d.tr.dest
 		} else if nsubj, err := d.tr.transform(tts); err == nil {
 			ndest = nsubj
@@ -948,7 +966,7 @@ func (a *Account) removeClient(c *client) int {
 
 func setExportAuth(ea *exportAuth, subject string, accounts []*Account, accountPos uint) error {
 	if accountPos > 0 {
-		token := strings.Split(subject, ".")
+		token := strings.Split(subject, tsep)
 		if len(token) < int(accountPos) || token[accountPos-1] != "*" {
 			return ErrInvalidSubject
 		}
@@ -2442,7 +2460,7 @@ func (a *Account) AddStreamExport(subject string, accounts []*Account) error {
 // AddStreamExport will add an export to the account. If accounts is nil
 // it will signify a public export, meaning anyone can import.
 // if accountPos is > 0, all imports will be granted where the following holds:
-// strings.Split(subject, ".")[accountPos] == account id will be granted.
+// strings.Split(subject, tsep)[accountPos] == account id will be granted.
 func (a *Account) addStreamExportWithAccountPos(subject string, accounts []*Account, accountPos uint) error {
 	if a == nil {
 		return ErrMissingAccount
@@ -4166,11 +4184,13 @@ func (dr *CacheDirAccResolver) Reload() error {
 // These can also be used for proper mapping on wildcard exports/imports.
 // These will be grouped and caching and locking are assumed to be in the upper layers.
 type transform struct {
-	src, dest string
-	dtoks     []string
-	stoks     []string
-	dtpi      [][]int // destination token position indexes
-	dtpinp    []int32 // destination token position index number of partitions
+	src, dest            string
+	dtoks                []string // destination tokens
+	stoks                []string // source tokens
+	dtokmftypes          []int16  // destination token mapping function types
+	dtokmftokindexesargs [][]int  // destination token mapping function array of source token index arguments
+	dtokmfintargs        []int32  // destination token mapping function int32 arguments
+	dtokmfstringargs     []string // destination token mapping function string arguments
 }
 
 func getMappingFunctionArgs(functionRegEx *regexp.Regexp, token string) []string {
@@ -4181,17 +4201,40 @@ func getMappingFunctionArgs(functionRegEx *regexp.Regexp, token string) []string
 	return nil
 }
 
-// Helper to pull raw place holder indexes and number of partitions. Returns -1 if not a place holder.
-func placeHolderIndex(token string) ([]int, int32, error) {
+// Helper for mapping functions that take a wildcard index and an integer as arguments
+func transformIndexIntArgsHelper(token string, args []string, transformType int16) (int16, []int, int32, string, error) {
+	if len(args) < 2 {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+	}
+	if len(args) > 2 {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
+	}
+	i, err := strconv.Atoi(strings.Trim(args[0], " "))
+	if err != nil {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+	}
+	mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[1], " "))
+	if err != nil {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+	}
+
+	return transformType, []int{i}, int32(mappingFunctionIntArg), _EMPTY_, nil
+}
+
+// Helper to ingest and index the transform destination token (e.g. $x or {{}}) in the token
+// returns a transformation type, and three function arguments: an array of source subject token indexes, and a single number (e.g. number of partitions, or a slice size), and a string (e.g.a split delimiter)
+
+func indexPlaceHolders(token string) (int16, []int, int32, string, error) {
 	length := len(token)
 	if length > 1 {
 		// old $1, $2, etc... mapping format still supported to maintain backwards compatibility
 		if token[0] == '$' { // simple non-partition mapping
 			tp, err := strconv.Atoi(token[1:])
 			if err != nil {
-				return []int{-1}, -1, nil
+				// other things rely on tokens starting with $ so not an error just leave it as is
+				return NoTransform, []int{-1}, -1, _EMPTY_, nil
 			}
-			return []int{tp}, -1, nil
+			return Wildcard, []int{tp}, -1, _EMPTY_, nil
 		}
 
 		// New 'mustache' style mapping
@@ -4200,45 +4243,92 @@ func placeHolderIndex(token string) ([]int, int32, error) {
 			args := getMappingFunctionArgs(wildcardMappingFunctionRegEx, token)
 			if args != nil {
 				if len(args) == 1 && args[0] == _EMPTY_ {
-					return []int{-1}, -1, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
 				}
 				if len(args) == 1 {
-					tp, err := strconv.Atoi(strings.Trim(args[0], " "))
+					tokenIndex, err := strconv.Atoi(strings.Trim(args[0], " "))
 					if err != nil {
-						return []int{}, -1, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
 					}
-					return []int{tp}, -1, nil
+					return Wildcard, []int{tokenIndex}, -1, _EMPTY_, nil
 				} else {
-					return []int{}, -1, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
 				}
 			}
+
 			// partition(number of partitions, token1, token2, ...)
 			args = getMappingFunctionArgs(partitionMappingFunctionRegEx, token)
 			if args != nil {
 				if len(args) < 2 {
-					return []int{-1}, -1, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
 				}
 				if len(args) >= 2 {
-					tphnp, err := strconv.Atoi(strings.Trim(args[0], " "))
+					mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[0], " "))
 					if err != nil {
-						return []int{}, -1, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
 					}
 					var numPositions = len(args[1:])
-					tps := make([]int, numPositions)
+					tokenIndexes := make([]int, numPositions)
 					for ti, t := range args[1:] {
 						i, err := strconv.Atoi(strings.Trim(t, " "))
 						if err != nil {
-							return []int{}, -1, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+							return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
 						}
-						tps[ti] = i
+						tokenIndexes[ti] = i
 					}
-					return tps, int32(tphnp), nil
+
+					return Partition, tokenIndexes, int32(mappingFunctionIntArg), _EMPTY_, nil
+				}
+			}
+
+			// SplitFromLeft(token, position)
+			args = getMappingFunctionArgs(splitFromLeftMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SplitFromLeft)
+			}
+
+			// SplitFromRight(token, position)
+			args = getMappingFunctionArgs(splitFromRightMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SplitFromRight)
+			}
+
+			// SliceFromLeft(token, position)
+			args = getMappingFunctionArgs(sliceFromLeftMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SliceFromLeft)
+			}
+
+			// SliceFromRight(token, position)
+			args = getMappingFunctionArgs(sliceFromRightMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SliceFromRight)
+			}
+
+			// split(token, deliminator)
+			args = getMappingFunctionArgs(splitMappingFunctionRegEx, token)
+			if args != nil {
+				if len(args) < 2 {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+				}
+				if len(args) > 2 {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
+				}
+				i, err := strconv.Atoi(strings.Trim(args[0], " "))
+				if err != nil {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+				}
+				if strings.Contains(args[1], " ") || strings.Contains(args[1], tsep) {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token: token, err: ErrorMappingDestinationFunctionInvalidArgument}
 				}
+
+				return Split, []int{i}, -1, args[1], nil
 			}
-			return []int{}, -1, &mappingDestinationErr{token, ErrUnknownMappingDestinationFunction}
+
+			return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrUnknownMappingDestinationFunction}
 		}
 	}
-	return []int{-1}, -1, nil
+	return NoTransform, []int{-1}, -1, _EMPTY_, nil
 }
 
 // SubjectTransformer transforms subjects using mappings
@@ -4266,8 +4356,10 @@ func newTransform(src, dest string) (*transform, error) {
 		return nil, ErrBadSubject
 	}
 
-	var dtpi [][]int
-	var dtpinb []int32
+	var dtokMappingFunctionTypes []int16
+	var dtokMappingFunctionTokenIndexes [][]int
+	var dtokMappingFunctionIntArgs []int32
+	var dtokMappingFunctionStringArgs []string
 
 	// If the src has partial wildcards then the dest needs to have the token place markers.
 	if npwcs > 0 || hasFwc {
@@ -4281,25 +4373,31 @@ func newTransform(src, dest string) (*transform, error) {
 
 		nphs := 0
 		for _, token := range dtokens {
-			tp, nb, err := placeHolderIndex(token)
+			tranformType, transformArgWildcardIndexes, transfomArgInt, transformArgString, err := indexPlaceHolders(token)
 			if err != nil {
 				return nil, err
 			}
-			if tp[0] >= 0 {
+
+			if tranformType == NoTransform {
+				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, NoTransform)
+				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, []int{-1})
+				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, -1)
+				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, _EMPTY_)
+			} else {
 				nphs++
 				// Now build up our runtime mapping from dest to source tokens.
 				var stis []int
-				for _, position := range tp {
-					if position > npwcs {
-						return nil, &mappingDestinationErr{fmt.Sprintf("%s: [%d]", token, position), ErrorMappingDestinationFunctionWildcardIndexOutOfRange}
+				for _, wildcardIndex := range transformArgWildcardIndexes {
+					if wildcardIndex > npwcs {
+						return nil, &mappingDestinationErr{fmt.Sprintf("%s: [%d]", token, wildcardIndex), ErrorMappingDestinationFunctionWildcardIndexOutOfRange}
 					}
-					stis = append(stis, sti[position])
+					stis = append(stis, sti[wildcardIndex])
 				}
-				dtpi = append(dtpi, stis)
-				dtpinb = append(dtpinb, nb)
-			} else {
-				dtpi = append(dtpi, []int{-1})
-				dtpinb = append(dtpinb, -1)
+				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, tranformType)
+				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, stis)
+				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, transfomArgInt)
+				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, transformArgString)
+
 			}
 		}
 		if nphs < npwcs {
@@ -4308,7 +4406,7 @@ func newTransform(src, dest string) (*transform, error) {
 		}
 	}
 
-	return &transform{src: src, dest: dest, dtoks: dtokens, stoks: stokens, dtpi: dtpi, dtpinp: dtpinb}, nil
+	return &transform{src: src, dest: dest, dtoks: dtokens, stoks: stokens, dtokmftypes: dtokMappingFunctionTypes, dtokmftokindexesargs: dtokMappingFunctionTokenIndexes, dtokmfintargs: dtokMappingFunctionIntArgs, dtokmfstringargs: dtokMappingFunctionStringArgs}, nil
 }
 
 // Match will take a literal published subject that is associated with a client and will match and transform
@@ -4370,41 +4468,110 @@ func (tr *transform) getHashPartition(key []byte, numBuckets int) string {
 
 // Do a transform on the subject to the dest subject.
 func (tr *transform) transform(tokens []string) (string, error) {
-	if len(tr.dtpi) == 0 {
+	if len(tr.dtokmftypes) == 0 {
 		return tr.dest, nil
 	}
 
 	var b strings.Builder
-	var token string
 
-	// We need to walk destination tokens and create the mapped subject pulling tokens from src.
+	// We need to walk destination tokens and create the mapped subject pulling tokens or mapping functions
 	// This is slow and that is ok, transforms should have caching layer in front for mapping transforms
 	// and export/import semantics with streams and services.
-	li := len(tr.dtpi) - 1
-	for i, index := range tr.dtpi {
-		// <0 means use destination token.
-		if index[0] < 0 {
-			token = tr.dtoks[i]
+	li := len(tr.dtokmftypes) - 1
+	for i, mfType := range tr.dtokmftypes {
+		if mfType == NoTransform {
 			// Break if fwc
-			if len(token) == 1 && token[0] == fwc {
+			if len(tr.dtoks[i]) == 1 && tr.dtoks[i][0] == fwc {
 				break
 			}
+			b.WriteString(tr.dtoks[i])
 		} else {
-			// >= 0 means use source map index to figure out which source token to pull.
-			if tr.dtpinp[i] > 0 { // there is a valid (i.e. not -1) value for number of partitions, this is a partition transform token
+			switch mfType {
+			case Partition:
 				var (
 					_buffer       [64]byte
 					keyForHashing = _buffer[:0]
 				)
-				for _, sourceToken := range tr.dtpi[i] {
+				for _, sourceToken := range tr.dtokmftokindexesargs[i] {
 					keyForHashing = append(keyForHashing, []byte(tokens[sourceToken])...)
 				}
-				token = tr.getHashPartition(keyForHashing, int(tr.dtpinp[i]))
-			} else { // back to normal substitution
-				token = tokens[tr.dtpi[i][0]]
+				b.WriteString(tr.getHashPartition(keyForHashing, int(tr.dtokmfintargs[i])))
+			case Wildcard: // simple substitution
+				b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]])
+			case SplitFromLeft:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				position := int(tr.dtokmfintargs[i])
+				if position > 0 && position < sourceTokenLen {
+					b.WriteString(sourceToken[:position])
+					b.WriteString(tsep)
+					b.WriteString(sourceToken[position:])
+				} else { // too small to split at the requested position: don't split
+					b.WriteString(sourceToken)
+				}
+			case SplitFromRight:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				position := int(tr.dtokmfintargs[i])
+				if position > 0 && position < sourceTokenLen {
+					b.WriteString(sourceToken[:sourceTokenLen-position])
+					b.WriteString(tsep)
+					b.WriteString(sourceToken[sourceTokenLen-position:])
+				} else { // too small to split at the requested position: don't split
+					b.WriteString(sourceToken)
+				}
+			case SliceFromLeft:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					for i := 0; i+sliceSize <= sourceTokenLen; i += sliceSize {
+						if i != 0 {
+							b.WriteString(tsep)
+						}
+						b.WriteString(sourceToken[i : i+sliceSize])
+						if i+sliceSize != sourceTokenLen && i+sliceSize+sliceSize > sourceTokenLen {
+							b.WriteString(tsep)
+							b.WriteString(sourceToken[i+sliceSize:])
+							break
+						}
+					}
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
+			case SliceFromRight:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					remainder := sourceTokenLen % sliceSize
+					if remainder > 0 {
+						b.WriteString(sourceToken[:remainder])
+						b.WriteString(tsep)
+					}
+					for i := remainder; i+sliceSize <= sourceTokenLen; i += sliceSize {
+						b.WriteString(sourceToken[i : i+sliceSize])
+						if i+sliceSize < sourceTokenLen {
+							b.WriteString(tsep)
+						}
+					}
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
+			case Split:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				splits := strings.Split(sourceToken, tr.dtokmfstringargs[i])
+				for j, split := range splits {
+					if split != _EMPTY_ {
+						b.WriteString(split)
+					}
+					if j < len(splits)-1 && splits[j+1] != _EMPTY_ && j != 0 {
+						b.WriteString(tsep)
+					}
+				}
 			}
 		}
-		b.WriteString(token)
+
 		if i < li {
 			b.WriteByte(btsep)
 		}
@@ -4424,7 +4591,7 @@ func (tr *transform) transform(tokens []string) (string, error) {
 
 // Reverse a transform.
 func (tr *transform) reverse() *transform {
-	if len(tr.dtpi) == 0 {
+	if len(tr.dtokmftokindexesargs) == 0 {
 		rtr, _ := newTransform(tr.dest, tr.src)
 		return rtr
 	}
diff --git a/server/const.go b/server/const.go
index f214552c..cf408566 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.4"
+	VERSION = "2.9.0-RC.6"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
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 64c2db53..35574cf5 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -161,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
 }
 
@@ -646,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))
@@ -742,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)
@@ -754,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)
 	}
 
@@ -1784,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]
 	}
 
@@ -5090,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.
@@ -5124,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
 }
@@ -5196,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 {
@@ -5249,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()
 	}
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_cluster.go b/server/jetstream_cluster.go
index 48195173..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
 }
 
@@ -5092,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/stream.go b/server/stream.go
index a7bb231a..16858ea2 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -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)
@@ -1482,31 +1497,6 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 	if mset.isLeader() && sendAdvisory {
 		mset.sendUpdateAdvisoryLocked()
 	}
-
-	// Check for AllowDirect
-	if mset.cfg.AllowDirect {
-		mset.subscribeToDirect()
-	} else if mset.directSub != nil {
-		mset.unsubscribe(mset.directSub)
-		mset.directSub = nil
-	}
-
-	// Check for mirror. If set but we are not a mirror just ignore for now.
-	if mset.cfg.MirrorDirect && mset.cfg.Mirror != nil {
-		if err := mset.subscribeToMirrorDirect(); err != nil {
-			// Disable since we had problems above.
-			mset.cfg.MirrorDirect = false
-		}
-	} else if !mset.cfg.MirrorDirect && mset.mirror != nil {
-		if mset.mirror.dsub != nil {
-			mset.unsubscribe(mset.mirror.dsub)
-			mset.mirror.dsub = nil
-		}
-		if mset.mirror.lbsub != nil {
-			mset.unsubscribe(mset.mirror.lbsub)
-			mset.mirror.lbsub = nil
-		}
-	}
 	mset.mu.Unlock()
 
 	if js != nil {
@@ -2311,7 +2301,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 +2344,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 +2355,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 +2386,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 +2415,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 +2453,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 +2479,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 +2499,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 +2599,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 +2927,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{})
 		}
 	}
 
diff --git a/server/sublist.go b/server/sublist.go
index 99230dde..bae47306 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -1140,7 +1140,13 @@ func ValidateMappingDestination(subject string) error {
 		}
 
 		if length > 4 && t[0] == '{' && t[1] == '{' && t[length-2] == '}' && t[length-1] == '}' {
-			if !partitionMappingFunctionRegEx.MatchString(t) && !wildcardMappingFunctionRegEx.MatchString(t) {
+			if !partitionMappingFunctionRegEx.MatchString(t) &&
+				!wildcardMappingFunctionRegEx.MatchString(t) &&
+				!splitFromLeftMappingFunctionRegEx.MatchString(t) &&
+				!splitFromRightMappingFunctionRegEx.MatchString(t) &&
+				!sliceFromLeftMappingFunctionRegEx.MatchString(t) &&
+				!sliceFromRightMappingFunctionRegEx.MatchString(t) &&
+				!splitMappingFunctionRegEx.MatchString(t) {
 				return &mappingDestinationErr{t, ErrUnknownMappingDestinationFunction}
 			} else {
 				continue


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 435b418b..f9bc8623 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -50,47 +50,54 @@ func simpleAccountServer(t *testing.T) (*Server, *Account, *Account) {
 
 func TestPlaceHolderIndex(t *testing.T) {
 	testString := "$1"
-	indexes, nbPartitions, err := placeHolderIndex(testString)
+	transformType, indexes, nbPartitions, _, err := indexPlaceHolders(testString)
+	var position int32
 
-	if err != nil || len(indexes) != 1 || indexes[0] != 1 || nbPartitions != -1 {
+	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 1 || nbPartitions != -1 {
 		t.Fatalf("Error parsing %s", testString)
 	}
 
 	testString = "{{partition(10,1,2,3)}}"
 
-	indexes, nbPartitions, err = placeHolderIndex(testString)
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
 
-	if err != nil || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
+	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
 		t.Fatalf("Error parsing %s", testString)
 	}
 
-	testString = "{{ partition(10,1,2,3) }}"
+	testString = "{{ Partition (10,1,2,3) }}"
 
-	indexes, nbPartitions, err = placeHolderIndex(testString)
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
 
-	if err != nil || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
+	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
 		t.Fatalf("Error parsing %s", testString)
 	}
 
-	testString = "{{partition (10,1,2,3)}}"
+	testString = "{{wildcard(2)}}"
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
 
-	indexes, nbPartitions, err = placeHolderIndex(testString)
+	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
+		t.Fatalf("Error parsing %s", testString)
+	}
 
-	if err != nil || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
+	testString = "{{SplitFromLeft(2,1)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != SplitFromLeft || len(indexes) != 1 || indexes[0] != 2 || position != 1 {
 		t.Fatalf("Error parsing %s", testString)
 	}
 
-	testString = "{{wildcard(2)}}"
-	indexes, nbPartitions, err = placeHolderIndex(testString)
+	testString = "{{SplitFromRight(3,2)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
 
-	if err != nil || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
+	if err != nil || transformType != SplitFromRight || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
 		t.Fatalf("Error parsing %s", testString)
 	}
 
-	testString = "{{ wildcard (2) }}"
-	indexes, nbPartitions, err = placeHolderIndex(testString)
+	testString = "{{SliceFromLeft(2,2)}}"
+	transformType, indexes, sliceSize, _, err := indexPlaceHolders(testString)
 
-	if err != nil || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
+	if err != nil || transformType != SliceFromLeft || len(indexes) != 1 || indexes[0] != 2 || sliceSize != 2 {
 		t.Fatalf("Error parsing %s", testString)
 	}
 }
@@ -3290,6 +3297,7 @@ func TestSubjectTransforms(t *testing.T) {
 	shouldErr("foo.*", "foo.{{wildcard()}}")       // Not enough arguments passed to the mapping function
 	shouldErr("foo.*", "foo.{{wildcard(1,2)}}")    // Too many arguments passed to the mapping function
 	shouldErr("foo.*", "foo.{{ wildcard5) }}")     // Bad mapping function
+	shouldErr("foo.*", "foo.{{splitLeft(2,2}}")    // arg out of range
 
 	shouldBeOK := func(src, dest string) *transform {
 		t.Helper()
@@ -3303,6 +3311,7 @@ func TestSubjectTransforms(t *testing.T) {
 	shouldBeOK("foo", "bar")
 	shouldBeOK("foo.*.bar.*.baz", "req.$2.$1")
 	shouldBeOK("baz.>", "mybaz.>")
+	shouldBeOK("*", "{{splitfromleft(1,1)}}")
 
 	shouldMatch := func(src, dest, sample, expected string) {
 		t.Helper()
@@ -3321,6 +3330,12 @@ func TestSubjectTransforms(t *testing.T) {
 	shouldMatch("baz.>", "my.pre.>", "baz.1.2.3", "my.pre.1.2.3")
 	shouldMatch("baz.>", "foo.bar.>", "baz.1.2.3", "foo.bar.1.2.3")
 	shouldMatch("*", "foo.bar.$1", "foo", "foo.bar.foo")
+	shouldMatch("*", "{{splitfromleft(1,3)}}", "12345", "123.45")
+	shouldMatch("*", "{{SplitFromRight(1,3)}}", "12345", "12.345")
+	shouldMatch("*", "{{SliceFromLeft(1,3)}}", "1234567890", "123.456.789.0")
+	shouldMatch("*", "{{SliceFromRight(1,3)}}", "1234567890", "1.234.567.890")
+	shouldMatch("*", "{{split(1,-)}}", "-abc-def--ghi-", "abc.def.ghi")
+	shouldMatch("*.*", "{{split(2,-)}}.{{splitfromleft(1,2)}}", "foo.-abc-def--ghij-", "abc.def.ghij.fo.o") // combo + checks split for multiple instance of deliminator and deliminator being at the start or end
 }
 
 func TestAccountSystemPermsWithGlobalAccess(t *testing.T) {
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 33db1505..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)
@@ -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()
@@ -4366,3 +4378,48 @@ func TestFileStoreEncryptedAES(t *testing.T) {
 		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_kv_test.go b/server/jetstream_chaos_kv_test.go
new file mode 100644
index 00000000..2b3f57e3
--- /dev/null
+++ b/server/jetstream_chaos_kv_test.go
@@ -0,0 +1,454 @@
+// 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 (
+	"fmt"
+	"math/rand"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+const (
+	chaosKvTestsClusterName = "KV_CHAOS_TEST"
+	chaosKvTestsBucketName  = "KV_CHAOS_TEST_BUCKET"
+	chaosKvTestsSubject     = "foo"
+	chaosKvTestsDebug       = false
+)
+
+// Creates KV store (a.k.a. bucket).
+func createBucketForKvChaosTest(t *testing.T, c *cluster, replicas int) {
+	t.Helper()
+
+	pubNc, pubJs := jsClientConnectCluster(t, c)
+	defer pubNc.Close()
+
+	config := nats.KeyValueConfig{
+		Bucket:      chaosKvTestsBucketName,
+		Replicas:    replicas,
+		Description: "Test bucket",
+	}
+
+	kvs, err := pubJs.CreateKeyValue(&config)
+	if err != nil {
+		t.Fatalf("Error creating bucket: %v", err)
+	}
+
+	status, err := kvs.Status()
+	if err != nil {
+		t.Fatalf("Error retrieving bucket status: %v", err)
+	}
+	t.Logf("Bucket created: %s", status.Bucket())
+}
+
+// Single client performs a set of PUT on a single key.
+// If PUT is successful, perform a GET on the same key.
+// If GET is successful, ensure key revision and value match the most recent successful write.
+func TestJetStreamChaosKvPutGet(t *testing.T) {
+
+	const numOps = 100_000
+	const clusterSize = 3
+	const replicas = 3
+	const key = "key"
+	const staleReadsOk = true // Set to false to check for violations of 'read committed' consistency
+
+	c := createJetStreamClusterExplicit(t, chaosKvTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createBucketForKvChaosTest(t, c, replicas)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: clusterSize, // Whole cluster outage
+			maxDownServers: clusterSize,
+			pause:          1 * time.Second,
+		},
+	)
+
+	nc, js := jsClientConnectCluster(t, c)
+	defer nc.Close()
+
+	// Create KV bucket
+	kv, err := js.KeyValue(chaosKvTestsBucketName)
+	if err != nil {
+		t.Fatalf("Failed to get KV store: %v", err)
+	}
+
+	// Initialize the only key
+	firstRevision, err := kv.Create(key, []byte("INITIAL VALUE"))
+	if err != nil {
+		t.Fatalf("Failed to create key: %v", err)
+	} else if firstRevision != 1 {
+		t.Fatalf("Unexpected revision: %d", firstRevision)
+	}
+
+	// Start chaos
+	chaos.start()
+	defer chaos.stop()
+
+	staleReadsCount := uint64(0)
+	successCount := uint64(0)
+
+	previousRevision := firstRevision
+
+putGetLoop:
+	for i := 1; i <= numOps; i++ {
+
+		if i%1000 == 0 {
+			t.Logf("Completed %d/%d PUT+GET operations", i, numOps)
+		}
+
+		// PUT a value
+		putValue := fmt.Sprintf("value-%d", i)
+		putRevision, err := kv.Put(key, []byte(putValue))
+		if err != nil {
+			t.Logf("PUT error: %v", err)
+			continue putGetLoop
+		}
+
+		// Check revision is monotonically increasing
+		if putRevision <= previousRevision {
+			t.Fatalf("PUT produced revision %d which is not greater than the previous successful PUT revision: %d", putRevision, previousRevision)
+		}
+
+		previousRevision = putRevision
+
+		// If PUT was successful, GET the same
+		kve, err := kv.Get(key)
+		if err == nats.ErrKeyNotFound {
+			t.Fatalf("GET key not found, but key does exists (last PUT revision: %d)", putRevision)
+		} else if err != nil {
+			t.Logf("GET error: %v", err)
+			continue putGetLoop
+		}
+
+		getValue := string(kve.Value())
+		getRevision := kve.Revision()
+
+		if putRevision > getRevision {
+			// Stale read, violates 'read committed' consistency criteria
+			if !staleReadsOk {
+				t.Fatalf("PUT value %s (rev: %d) then read value %s (rev: %d)", putValue, putRevision, getValue, getRevision)
+			} else {
+				staleReadsCount += 1
+			}
+		} else if putRevision < getRevision {
+			// Returned revision is higher than any ever written, this should never happen
+			t.Fatalf("GET returned revision %d, but most recent expected revision is %d", getRevision, putRevision)
+		} else if putValue != getValue {
+			// Returned revision matches latest, but values do not, this should never happen
+			t.Fatalf("GET returned revision %d with value %s, but value %s was just committed for that same revision", getRevision, getValue, putValue)
+		} else {
+			// Get returned the latest revision/value
+			successCount += 1
+			if chaosKvTestsDebug {
+				t.Logf("PUT+GET %s=%s (rev: %d)", key, putValue, putRevision)
+			}
+		}
+	}
+
+	t.Logf("Completed %d PUT+GET cycles of which %d successful, %d GETs returned a stale value", numOps, successCount, staleReadsCount)
+}
+
+// A variant TestJetStreamChaosKvPutGet where PUT is retried until successful, and GET is retried until it returns the latest known key revision.
+// This validates than a confirmed PUT value is never lost, and becomes eventually visible.
+func TestJetStreamChaosKvPutGetWithRetries(t *testing.T) {
+
+	const numOps = 10_000
+	const maxRetries = 20
+	const retryDelay = 100 * time.Millisecond
+	const clusterSize = 3
+	const replicas = 3
+	const key = "key"
+
+	c := createJetStreamClusterExplicit(t, chaosKvTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createBucketForKvChaosTest(t, c, replicas)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: clusterSize, // Whole cluster outage
+			maxDownServers: clusterSize,
+			pause:          1 * time.Second,
+		},
+	)
+
+	nc, js := jsClientConnectCluster(t, c)
+	defer nc.Close()
+
+	kv, err := js.KeyValue(chaosKvTestsBucketName)
+	if err != nil {
+		t.Fatalf("Failed to get KV store: %v", err)
+	}
+
+	// Initialize key value
+	firstRevision, err := kv.Create(key, []byte("INITIAL VALUE"))
+	if err != nil {
+		t.Fatalf("Failed to create key: %v", err)
+	} else if firstRevision != 1 {
+		t.Fatalf("Unexpected revision: %d", firstRevision)
+	}
+
+	// Start chaos
+	chaos.start()
+	defer chaos.stop()
+
+	staleReadCount := 0
+	previousRevision := firstRevision
+
+putGetLoop:
+	for i := 1; i <= numOps; i++ {
+
+		if i%1000 == 0 {
+			t.Logf("Completed %d/%d PUT+GET operations", i, numOps)
+		}
+
+		putValue := fmt.Sprintf("value-%d", i)
+		putRevision := uint64(0)
+
+		// Put new value for key, retry until successful or out of retries
+	putRetryLoop:
+		for r := 0; r <= maxRetries; r++ {
+			var putErr error
+			putRevision, putErr = kv.Put(key, []byte(putValue))
+			if putErr == nil {
+				break putRetryLoop
+			} else if r == maxRetries {
+				t.Fatalf("Failed to PUT (retried %d times): %v", maxRetries, putErr)
+			} else {
+				if chaosKvTestsDebug {
+					t.Logf("PUT error: %v", putErr)
+				}
+				time.Sleep(retryDelay)
+			}
+		}
+
+		// Ensure key version is monotonically increasing
+		if putRevision <= previousRevision {
+			t.Fatalf("Latest PUT created revision %d which is not greater than the previous revision: %d", putRevision, previousRevision)
+		}
+		previousRevision = putRevision
+
+		// Read value for key, retry until successful, and validate corresponding version and value
+	getRetryLoop:
+		for r := 0; r <= maxRetries; r++ {
+			var getErr error
+			kve, getErr := kv.Get(key)
+			if getErr != nil && r == maxRetries {
+				t.Fatalf("Failed to GET (retried %d times): %v", maxRetries, getErr)
+			} else if getErr != nil {
+				if chaosKvTestsDebug {
+					t.Logf("GET error: %v", getErr)
+				}
+				time.Sleep(retryDelay)
+				continue getRetryLoop
+			}
+
+			// GET successful, check revision and value
+			getValue := string(kve.Value())
+			getRevision := kve.Revision()
+
+			if putRevision == getRevision {
+				if putValue != getValue {
+					t.Fatalf("Unexpected value %s for revision %d, expected: %s", getValue, getRevision, putValue)
+				}
+				if chaosKvTestsDebug {
+					t.Logf("PUT+GET %s=%s (rev: %d) (retry: %d)", key, putValue, putRevision, r)
+				}
+				continue putGetLoop
+			} else if getRevision > putRevision {
+				t.Fatalf("GET returned version that should not exist yet: %d, last created: %d", getRevision, putRevision)
+			} else { // get revision < put revision
+				staleReadCount += 1
+				if chaosKvTestsDebug {
+					t.Logf("GET got stale value: %v (rev: %d, latest: %d)", getValue, getRevision, putRevision)
+				}
+				time.Sleep(retryDelay)
+				continue getRetryLoop
+			}
+		}
+	}
+
+	t.Logf("Client completed %d PUT+GET cycles, %d GET returned a stale value", numOps, staleReadCount)
+}
+
+// Multiple clients updating a finite set of keys with CAS semantics.
+// TODO check that revision is never lower than last one seen
+// TODO check that KeyNotFound is never returned, as keys are initialized beforehand
+func TestJetStreamChaosKvCAS(t *testing.T) {
+	const numOps = 10_000
+	const maxRetries = 50
+	const retryDelay = 300 * time.Millisecond
+	const clusterSize = 3
+	const replicas = 3
+	const numKeys = 15
+	const numClients = 5
+
+	c := createJetStreamClusterExplicit(t, chaosKvTestsClusterName, clusterSize)
+	defer c.shutdown()
+
+	createBucketForKvChaosTest(t, c, replicas)
+
+	chaos := createClusterChaosMonkeyController(
+		t,
+		c,
+		&clusterBouncerChaosMonkey{
+			minDowntime:    0 * time.Second,
+			maxDowntime:    2 * time.Second,
+			minDownServers: clusterSize, // Whole cluster outage
+			maxDownServers: clusterSize,
+			pause:          1 * time.Second,
+		},
+	)
+
+	nc, js := jsClientConnectCluster(t, c)
+	defer nc.Close()
+
+	// Create bucket
+	kv, err := js.KeyValue(chaosKvTestsBucketName)
+	if err != nil {
+		t.Fatalf("Failed to get KV store: %v", err)
+	}
+
+	// Create set of keys and initialize them with dummy value
+	keys := make([]string, numKeys)
+	for k := 0; k < numKeys; k++ {
+		key := fmt.Sprintf("key-%d", k)
+		keys[k] = key
+
+		_, err := kv.Create(key, []byte("Initial value"))
+		if err != nil {
+			t.Fatalf("Failed to create key: %v", err)
+		}
+	}
+
+	wgStart := sync.WaitGroup{}
+	wgComplete := sync.WaitGroup{}
+
+	// Client routine
+	client := func(clientId int, kv nats.KeyValue) {
+		defer wgComplete.Done()
+
+		rng := rand.New(rand.NewSource(int64(clientId)))
+		successfulUpdates := 0
+		casRejectUpdates := 0
+		otherUpdateErrors := 0
+
+		// Map to track last known revision for each of the keys
+		knownRevisions := map[string]uint64{}
+		for _, key := range keys {
+			knownRevisions[key] = 0
+		}
+
+		// Wait for all clients to reach this point before proceeding
+		wgStart.Done()
+		wgStart.Wait()
+
+		for i := 1; i <= numOps; i++ {
+
+			if i%1000 == 0 {
+				t.Logf("Client %d completed %d/%d updates", clientId, i, numOps)
+			}
+
+			// Pick random key from the set
+			key := keys[rng.Intn(numKeys)]
+
+			// Prepare unique value to be written
+			value := fmt.Sprintf("client: %d operation %d", clientId, i)
+
+			// Try to update a key with CAS
+			newRevision, updateErr := kv.Update(key, []byte(value), knownRevisions[key])
+			if updateErr == nil {
+				// Update successful
+				knownRevisions[key] = newRevision
+				successfulUpdates += 1
+				if chaosKvTestsDebug {
+					t.Logf("Client %d updated key %s, new revision: %d", clientId, key, newRevision)
+				}
+			} else if updateErr != nil && strings.Contains(fmt.Sprint(updateErr), "wrong last sequence") {
+				// CAS rejected update, learn current revision for this key
+				casRejectUpdates += 1
+			getRetryLoop:
+				for r := 0; r <= maxRetries; r++ {
+					kve, getErr := kv.Get(key)
+					if getErr == nil {
+						currentRevision := kve.Revision()
+						if currentRevision < knownRevisions[key] {
+							// Revision number moved backward, this should never happen
+							t.Fatalf("Current revision for key %s is %d, which is lower than the last known revision %d", key, currentRevision, knownRevisions[key])
+						}
+						knownRevisions[key] = currentRevision
+						if chaosKvTestsDebug {
+							t.Logf("Client %d learn key %s revision: %d", clientId, key, currentRevision)
+						}
+						break getRetryLoop
+					} else if r == maxRetries {
+						t.Fatalf("Failed to GET (retried %d times): %v", maxRetries, getErr)
+					} else {
+						time.Sleep(retryDelay)
+					}
+				}
+			} else {
+				// Other update error
+				otherUpdateErrors += 1
+				if chaosKvTestsDebug {
+					t.Logf("Client %d update error for key %s: %v", clientId, key, updateErr)
+				}
+				time.Sleep(retryDelay)
+			}
+		}
+		t.Logf("Client %d done, %d kv updates, %d CAS rejected, %d other errors", clientId, successfulUpdates, casRejectUpdates, otherUpdateErrors)
+	}
+
+	// Launch all clients
+	for i := 1; i <= numClients; i++ {
+		cNc, cJs := jsClientConnectCluster(t, c)
+		defer cNc.Close()
+
+		cKv, err := cJs.KeyValue(chaosKvTestsBucketName)
+		if err != nil {
+			t.Fatalf("Failed to get KV store: %v", err)
+		}
+
+		wgStart.Add(1)
+		wgComplete.Add(1)
+		go client(i, cKv)
+	}
+
+	// Wait for clients to be connected and ready
+	wgStart.Wait()
+
+	// Start failures
+	chaos.start()
+	defer chaos.stop()
+
+	// Wait for all clients to be done
+	wgComplete.Wait()
+}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 0f17abc9..8598a3d9 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) {
@@ -12818,3 +12831,76 @@ func TestJetStreamClusterLeaderAbortsCatchupOnFollowerError(t *testing.T) {
 		t.Fatal("Expected err response from the remote")
 	}
 }
+
+func TestJetStreamClusterStreamDirectGetNotTooSoon(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3F", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomServer()
+	nc, _ := jsClientConnect(t, s)
+	defer nc.Close()
+
+	// Do by hand for now.
+	cfg := &StreamConfig{
+		Name:        "TEST",
+		Storage:     FileStorage,
+		Subjects:    []string{"foo"},
+		Replicas:    3,
+		MaxMsgsPer:  1,
+		AllowDirect: true,
+	}
+	addStream(t, nc, cfg)
+	sendStreamMsg(t, nc, "foo", "bar")
+
+	getSubj := fmt.Sprintf(JSDirectGetLastBySubjectT, "TEST", "foo")
+
+	// Make sure we get all direct subs.
+	checkForDirectSubs := func() {
+		t.Helper()
+		checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
+			for _, s := range c.servers {
+				mset, err := s.GlobalAccount().lookupStream("TEST")
+				if err != nil {
+					return err
+				}
+				mset.mu.RLock()
+				hasBoth := mset.directSub != nil && mset.lastBySub != nil
+				mset.mu.RUnlock()
+				if !hasBoth {
+					return fmt.Errorf("%v does not have both direct subs registered", s)
+				}
+			}
+			return nil
+		})
+	}
+
+	_, err := nc.Request(getSubj, nil, time.Second)
+	require_NoError(t, err)
+
+	checkForDirectSubs()
+
+	// We want to make sure that when starting up we do not listen until we have a leader.
+	nc.Close()
+	c.stopAll()
+
+	// Start just one..
+	s, opts := RunServerWithConfig(c.opts[0].ConfigFile)
+	c.servers[0] = s
+	c.opts[0] = opts
+
+	nc, _ = jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err = nc.Request(getSubj, nil, time.Second)
+	require_Error(t, err, nats.ErrTimeout)
+
+	// Now start all and make sure they all eventually have subs for direct access.
+	c.restartAll()
+	c.waitOnStreamLeader("$G", "TEST")
+
+	_, err = nc.Request(getSubj, nil, time.Second)
+	require_NoError(t, err)
+
+	checkForDirectSubs()
+}
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/sublist_test.go b/server/sublist_test.go
index 05a14dd2..f1fb1fde 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -669,6 +669,8 @@ func TestValidateDestinationSubject(t *testing.T) {
 	checkError(ValidateMappingDestination("foo.{{ wildcard(1) }}"), nil, t)
 	checkError(ValidateMappingDestination("foo.{{wildcard( 1 )}}"), nil, t)
 	checkError(ValidateMappingDestination("foo.{{partition(2,1)}}"), nil, t)
+	checkError(ValidateMappingDestination("foo.{{SplitFromLeft(2,1)}}"), nil, t)
+	checkError(ValidateMappingDestination("foo.{{SplitFromRight(2,1)}}"), nil, t)
 	checkError(ValidateMappingDestination("foo.{{unknown(1)}}"), ErrInvalidMappingDestination, t)
 	checkError(ValidateMappingDestination("foo..}"), ErrInvalidMappingDestination, t)
 	checkError(ValidateMappingDestination("foo. bar}"), ErrInvalidMappingDestinationSubject, t)