Repository: nats-io/nats-server
Pull request #3822 (2023-01-26 21:19:33)

Add batch completed status to Pull Consumers
Signed-off-by: Tomasz Pietrek <tomasz@nats.io>

This adds new status requested by #3816 

It is consistent with other statuses with pending data headers.
It can break simplification clients, or thos exposing full `Request` config (max_bytes and batch).

That is reflected in the need of test adoption.


/cc @nats-io/core


Jarema: @aricart FYI, this should fix issue of empty batch + max_bytes without any information.
derekcollison: Check build and make sure green though..

Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 5bd07472..2d68641b 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -18,6 +18,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"hash/maphash"
 	"io"
 	"math"
@@ -25,6 +26,7 @@ import (
 	"net/http"
 	"net/textproto"
 	"reflect"
+	"regexp"
 	"sort"
 	"strconv"
 	"strings"
@@ -113,8 +115,8 @@ type streamImport struct {
 	acc     *Account
 	from    string
 	to      string
-	tr      *subjectTransform
-	rtr     *subjectTransform
+	tr      *transform
+	rtr     *transform
 	claim   *jwt.Import
 	usePub  bool
 	invalid bool
@@ -130,7 +132,7 @@ type serviceImport struct {
 	sid         []byte
 	from        string
 	to          string
-	tr          *subjectTransform
+	tr          *transform
 	ts          int64
 	rt          ServiceRespType
 	latency     *serviceLatency
@@ -164,6 +166,28 @@ const (
 	Chunked
 )
 
+var commaSeparatorRegEx = regexp.MustCompile(`,\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 {
 	switch rt {
@@ -566,7 +590,7 @@ func NewMapDest(subject string, weight uint8) *MapDest {
 
 // destination is for internal representation for a weighted mapped destination.
 type destination struct {
-	tr     *subjectTransform
+	tr     *transform
 	weight uint8
 }
 
@@ -619,7 +643,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 		if err != nil {
 			return err
 		}
-		tr, err := newSubjectTransform(src, d.Subject)
+		tr, err := newTransform(src, d.Subject)
 		if err != nil {
 			return err
 		}
@@ -650,7 +674,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 				// We need to make the appropriate markers for the wildcards etc.
 				dest = transformTokenize(dest)
 			}
-			tr, err := newSubjectTransform(src, dest)
+			tr, err := newTransform(src, dest)
 			if err != nil {
 				return nil, err
 			}
@@ -706,6 +730,38 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 	return nil
 }
 
+// Helper function to tokenize subjects with partial wildcards into formal transform destinations.
+// e.g. foo.*.* -> foo.$1.$2
+func transformTokenize(subject string) string {
+	// We need to make the appropriate markers for the wildcards etc.
+	i := 1
+	var nda []string
+	for _, token := range strings.Split(subject, tsep) {
+		if token == "*" {
+			nda = append(nda, fmt.Sprintf("$%d", i))
+			i++
+		} else {
+			nda = append(nda, token)
+		}
+	}
+	return strings.Join(nda, tsep)
+}
+
+func transformUntokenize(subject string) (string, []string) {
+	var phs []string
+	var nda []string
+
+	for _, token := range strings.Split(subject, tsep) {
+		if len(token) > 1 && token[0] == '$' && token[1] >= '1' && token[1] <= '9' {
+			phs = append(phs, token)
+			nda = append(nda, "*")
+		} else {
+			nda = append(nda, token)
+		}
+	}
+	return strings.Join(nda, tsep), phs
+}
+
 // RemoveMapping will remove an existing mapping.
 func (a *Account) RemoveMapping(src string) bool {
 	a.mu.Lock()
@@ -813,8 +869,8 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 	if d != nil {
 		if len(d.tr.dtokmftokindexesargs) == 0 {
 			ndest = d.tr.dest
-		} else {
-			ndest = d.tr.TransformTokenizedSubject(tts)
+		} else if nsubj, err := d.tr.transform(tts); err == nil {
+			ndest = nsubj
 		}
 	}
 
@@ -1818,7 +1874,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 	// Check to see if we have a wildcard
 	var (
 		usePub bool
-		tr     *subjectTransform
+		tr     *transform
 		err    error
 	)
 	if subjectHasWildcard(to) {
@@ -1828,7 +1884,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		} else {
 			to, _ = transformUntokenize(to)
 			// Create a transform. Do so in reverse such that $ symbols only exist in to
-			if tr, err = newSubjectTransform(to, transformTokenize(from)); err != nil {
+			if tr, err = newTransform(to, transformTokenize(from)); err != nil {
 				a.mu.Unlock()
 				return nil, fmt.Errorf("failed to create mapping transform for service import subject %q to %q: %v",
 					from, to, err)
@@ -2367,7 +2423,7 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 
 	var (
 		usePub bool
-		tr     *subjectTransform
+		tr     *transform
 		err    error
 	)
 	if subjectHasWildcard(from) {
@@ -2375,7 +2431,7 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 			usePub = true
 		} else {
 			// Create a transform
-			if tr, err = newSubjectTransform(from, transformTokenize(to)); err != nil {
+			if tr, err = newTransform(from, transformTokenize(to)); err != nil {
 				return fmt.Errorf("failed to create mapping transform for stream import subject %q to %q: %v",
 					from, to, err)
 			}
@@ -4230,3 +4286,439 @@ func (dr *CacheDirAccResolver) Start(s *Server) error {
 func (dr *CacheDirAccResolver) Reload() error {
 	return dr.DirAccResolver.Reload()
 }
+
+// Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters.
+// 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 // 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 {
+	commandStrings := functionRegEx.FindStringSubmatch(token)
+	if len(commandStrings) > 1 {
+		return commaSeparatorRegEx.Split(commandStrings[1], -1)
+	}
+	return nil
+}
+
+// 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 {
+				// other things rely on tokens starting with $ so not an error just leave it as is
+				return NoTransform, []int{-1}, -1, _EMPTY_, nil
+			}
+			return Wildcard, []int{tp}, -1, _EMPTY_, nil
+		}
+
+		// New 'mustache' style mapping
+		if length > 4 && token[0] == '{' && token[1] == '{' && token[length-2] == '}' && token[length-1] == '}' {
+			// wildcard(wildcard token index) (equivalent to $)
+			args := getMappingFunctionArgs(wildcardMappingFunctionRegEx, token)
+			if args != nil {
+				if len(args) == 1 && args[0] == _EMPTY_ {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+				}
+				if len(args) == 1 {
+					tokenIndex, err := strconv.Atoi(strings.Trim(args[0], " "))
+					if err != nil {
+						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+					}
+					return Wildcard, []int{tokenIndex}, -1, _EMPTY_, nil
+				} else {
+					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 BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+				}
+				if len(args) >= 2 {
+					mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[0], " "))
+					if err != nil {
+						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+					}
+					var numPositions = len(args[1:])
+					tokenIndexes := make([]int, numPositions)
+					for ti, t := range args[1:] {
+						i, err := strconv.Atoi(strings.Trim(t, " "))
+						if err != nil {
+							return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+						}
+						tokenIndexes[ti] = i
+					}
+
+					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 BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrUnknownMappingDestinationFunction}
+		}
+	}
+	return NoTransform, []int{-1}, -1, _EMPTY_, nil
+}
+
+// SubjectTransformer transforms subjects using mappings
+//
+// This API is not part of the public API and not subject to SemVer protections
+type SubjectTransformer interface {
+	Match(string) (string, error)
+}
+
+// NewSubjectTransformer creates a new SubjectTransformer
+//
+// This API is not part of the public API and not subject to SemVer protections
+func NewSubjectTransformer(src, dest string) (SubjectTransformer, error) {
+	return newTransform(src, dest)
+}
+
+// newTransform will create a new transform checking the src and dest subjects for accuracy.
+func newTransform(src, dest string) (*transform, error) {
+	// Both entries need to be valid subjects.
+	sv, stokens, npwcs, hasFwc := subjectInfo(src)
+	dv, dtokens, dnpwcs, dHasFwc := subjectInfo(dest)
+
+	// Make sure both are valid, match fwc if present and there are no pwcs in the dest subject.
+	if !sv || !dv || dnpwcs > 0 || hasFwc != dHasFwc {
+		return nil, ErrBadSubject
+	}
+
+	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 {
+		// We need to count to make sure that the dest has token holders for the pwcs.
+		sti := make(map[int]int)
+		for i, token := range stokens {
+			if len(token) == 1 && token[0] == pwc {
+				sti[len(sti)+1] = i
+			}
+		}
+
+		nphs := 0
+		for _, token := range dtokens {
+			tranformType, transformArgWildcardIndexes, transfomArgInt, transformArgString, err := indexPlaceHolders(token)
+			if err != nil {
+				return nil, err
+			}
+
+			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 _, wildcardIndex := range transformArgWildcardIndexes {
+					if wildcardIndex > npwcs {
+						return nil, &mappingDestinationErr{fmt.Sprintf("%s: [%d]", token, wildcardIndex), ErrorMappingDestinationFunctionWildcardIndexOutOfRange}
+					}
+					stis = append(stis, sti[wildcardIndex])
+				}
+				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, tranformType)
+				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, stis)
+				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, transfomArgInt)
+				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, transformArgString)
+
+			}
+		}
+		if nphs < npwcs {
+			// not all wildcards are being used in the destination
+			return nil, &mappingDestinationErr{dest, ErrMappingDestinationNotUsingAllWildcards}
+		}
+	}
+
+	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
+// the subject if possible.
+//
+// This API is not part of the public API and not subject to SemVer protections
+func (tr *transform) Match(subject string) (string, error) {
+	// TODO(dlc) - We could add in client here to allow for things like foo -> foo.$ACCOUNT
+
+	// Special case: matches any and no no-op transform. May not be legal config for some features
+	// but specific validations made at transform create time
+	if (tr.src == fwcs || tr.src == _EMPTY_) && (tr.dest == fwcs || tr.dest == _EMPTY_) {
+		return subject, nil
+	}
+
+	// Tokenize the subject. This should always be a literal subject.
+	tsa := [32]string{}
+	tts := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			tts = append(tts, subject[start:i])
+			start = i + 1
+		}
+	}
+	tts = append(tts, subject[start:])
+	if !isValidLiteralSubject(tts) {
+		return _EMPTY_, ErrBadSubject
+	}
+
+	if (tr.src == _EMPTY_ || tr.src == fwcs) || isSubsetMatch(tts, tr.src) {
+		return tr.transform(tts)
+	}
+	return _EMPTY_, ErrNoTransforms
+}
+
+// transformSubject do not need to match, just transform.
+func (tr *transform) transformSubject(subject string) (string, error) {
+	// Tokenize the subject.
+	tsa := [32]string{}
+	tts := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			tts = append(tts, subject[start:i])
+			start = i + 1
+		}
+	}
+	tts = append(tts, subject[start:])
+	return tr.transform(tts)
+}
+
+func (tr *transform) getHashPartition(key []byte, numBuckets int) string {
+	h := fnv.New32a()
+	h.Write(key)
+
+	return strconv.Itoa(int(h.Sum32() % uint32(numBuckets)))
+}
+
+// Do a transform on the subject to the dest subject.
+func (tr *transform) transform(tokens []string) (string, error) {
+	if len(tr.dtokmftypes) == 0 {
+		return tr.dest, nil
+	}
+
+	var b strings.Builder
+
+	// 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.dtokmftypes) - 1
+	for i, mfType := range tr.dtokmftypes {
+		if mfType == NoTransform {
+			// Break if fwc
+			if len(tr.dtoks[i]) == 1 && tr.dtoks[i][0] == fwc {
+				break
+			}
+			b.WriteString(tr.dtoks[i])
+		} else {
+			switch mfType {
+			case Partition:
+				var (
+					_buffer       [64]byte
+					keyForHashing = _buffer[:0]
+				)
+				for _, sourceToken := range tr.dtokmftokindexesargs[i] {
+					keyForHashing = append(keyForHashing, []byte(tokens[sourceToken])...)
+				}
+				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 && split == _EMPTY_) {
+						b.WriteString(tsep)
+					}
+				}
+			}
+		}
+
+		if i < li {
+			b.WriteByte(btsep)
+		}
+	}
+
+	// We may have more source tokens available. This happens with ">".
+	if tr.dtoks[len(tr.dtoks)-1] == ">" {
+		for sli, i := len(tokens)-1, len(tr.stoks)-1; i < len(tokens); i++ {
+			b.WriteString(tokens[i])
+			if i < sli {
+				b.WriteByte(btsep)
+			}
+		}
+	}
+	return b.String(), nil
+}
+
+// Reverse a transform.
+func (tr *transform) reverse() *transform {
+	if len(tr.dtokmftokindexesargs) == 0 {
+		rtr, _ := newTransform(tr.dest, tr.src)
+		return rtr
+	}
+	// If we are here we need to dynamically get the correct reverse
+	// of this transform.
+	nsrc, phs := transformUntokenize(tr.dest)
+	var nda []string
+	for _, token := range tr.stoks {
+		if token == "*" {
+			if len(phs) == 0 {
+				// TODO(dlc) - Should not happen
+				return nil
+			}
+			nda = append(nda, phs[0])
+			phs = phs[1:]
+		} else {
+			nda = append(nda, token)
+		}
+	}
+	ndest := strings.Join(nda, tsep)
+	rtr, _ := newTransform(nsrc, ndest)
+	return rtr
+}
diff --git a/server/client.go b/server/client.go
index 0c146114..798a89e2 100644
--- a/server/client.go
+++ b/server/client.go
@@ -2754,8 +2754,10 @@ func (c *client) addShadowSub(sub *subscription, ime *ime) (*subscription, error
 		if ime.overlapSubj != _EMPTY_ {
 			s = ime.overlapSubj
 		}
-		subj := im.rtr.TransformSubject(s)
-
+		subj, err := im.rtr.transformSubject(s)
+		if err != nil {
+			return nil, err
+		}
 		nsub.subject = []byte(subj)
 	} else if !im.usePub || (im.usePub && ime.overlapSubj != _EMPTY_) || !ime.dyn {
 		if ime.overlapSubj != _EMPTY_ {
@@ -3033,7 +3035,7 @@ func (c *client) msgHeaderForRouteOrLeaf(subj, reply []byte, rt *routeTarget, ac
 		// Remap subject if its a shadow subscription, treat like a normal client.
 		if rt.sub.im != nil {
 			if rt.sub.im.tr != nil {
-				to := rt.sub.im.tr.TransformSubject(string(subj))
+				to, _ := rt.sub.im.tr.transformSubject(string(subj))
 				subj = []byte(to)
 			} else if !rt.sub.im.usePub {
 				subj = []byte(rt.sub.im.to)
@@ -3989,7 +3991,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 
 	if si.tr != nil {
 		// FIXME(dlc) - This could be slow, may want to look at adding cache to bare transforms?
-		to = si.tr.TransformSubject(subject)
+		to, _ = si.tr.transformSubject(subject)
 	} else if si.usePub {
 		to = subject
 	}
@@ -4278,7 +4280,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 				continue
 			}
 			if sub.im.tr != nil {
-				to := sub.im.tr.TransformSubject(string(subject))
+				to, _ := sub.im.tr.transformSubject(string(subject))
 				dsubj = append(_dsubj[:0], to...)
 			} else if sub.im.usePub {
 				dsubj = append(_dsubj[:0], subj...)
@@ -4425,7 +4427,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 					continue
 				}
 				if sub.im.tr != nil {
-					to := sub.im.tr.TransformSubject(string(subject))
+					to, _ := sub.im.tr.transformSubject(string(subject))
 					dsubj = append(_dsubj[:0], to...)
 				} else if sub.im.usePub {
 					dsubj = append(_dsubj[:0], subj...)
diff --git a/server/consumer.go b/server/consumer.go
index 0ee07294..dd3ee7ad 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -38,6 +38,9 @@ const (
 	JSPullRequestPendingBytes = "Nats-Pending-Bytes"
 )
 
+// Headers sent when batch size was completed, but there were remaining bytes.
+const JsPullRequestRemainingBytesT = "NATS/1.0 409 Batch Completed\r\n%s: %d\r\n%s: %d\r\n\r\n"
+
 type ConsumerInfo struct {
 	Stream         string          `json:"stream_name"`
 	Name           string          `json:"name"`
@@ -466,8 +469,17 @@ func checkConsumerCfg(
 		}
 	}
 
-	if config.FilterSubject != _EMPTY_ && !IsValidSubject(config.FilterSubject) {
-		return NewJSStreamInvalidConfigError(ErrBadSubject)
+	// As best we can make sure the filtered subject is valid.
+	if config.FilterSubject != _EMPTY_ {
+		subjects := copyStrings(cfg.Subjects)
+		// explicitly skip validFilteredSubject when recovering
+		hasExt := isRecovering
+		if !isRecovering {
+			subjects, hasExt = gatherSourceMirrorSubjects(subjects, cfg, acc)
+		}
+		if !hasExt && !validFilteredSubject(config.FilterSubject, subjects) {
+			return NewJSConsumerFilterNotSubsetError()
+		}
 	}
 
 	// Helper function to formulate similar errors.
@@ -3215,6 +3227,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			ackReply string
 			delay    time.Duration
 			sz       int
+			wrn, wrb int
 		)
 		o.mu.Lock()
 		// consumer is closed when mset is set to nil.
@@ -3268,6 +3281,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		if o.isPushMode() {
 			dsubj = o.dsubj
 		} else if wr := o.nextWaiting(sz); wr != nil {
+			wrn, wrb = wr.n, wr.b
 			dsubj = wr.reply
 			if done := wr.recycleIfDone(); done && o.node != nil {
 				o.removeClusterPendingRequest(dsubj)
@@ -3321,6 +3335,10 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		// Do actual delivery.
 		o.deliverMsg(dsubj, ackReply, pmsg, dc, rp)
 
+		// If given request fulfilled batch size, but there are still pending bytes, send information about it.
+		if wrn <= 0 && wrb > 0 {
+			o.outq.send(newJSPubMsg(dsubj, _EMPTY_, _EMPTY_, []byte(fmt.Sprintf(JsPullRequestRemainingBytesT, JSPullRequestPendingMsgs, wrn, JSPullRequestPendingBytes, wrb)), nil, nil, 0))
+		}
 		// Reset our idle heartbeat timer if set.
 		if hb != nil {
 			hb.Reset(hbd)
@@ -4263,6 +4281,26 @@ func deliveryFormsCycle(cfg *StreamConfig, deliverySubject string) bool {
 	return false
 }
 
+// Check that the filtered subject is valid given a set of stream subjects.
+func validFilteredSubject(filteredSubject string, subjects []string) bool {
+	if !IsValidSubject(filteredSubject) {
+		return false
+	}
+	hasWC := subjectHasWildcard(filteredSubject)
+
+	for _, subject := range subjects {
+		if subjectIsSubsetMatch(filteredSubject, subject) {
+			return true
+		}
+		// If we have a wildcard as the filtered subject check to see if we are
+		// a wider scope but do match a subject.
+		if hasWC && subjectIsSubsetMatch(subject, filteredSubject) {
+			return true
+		}
+	}
+	return false
+}
+
 // switchToEphemeral is called on startup when recovering ephemerals.
 func (o *consumer) switchToEphemeral() {
 	o.mu.Lock()
diff --git a/server/stream.go b/server/stream.go
index 5d4226d1..7be7c203 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -58,9 +58,6 @@ type StreamConfig struct {
 	Mirror       *StreamSource   `json:"mirror,omitempty"`
 	Sources      []*StreamSource `json:"sources,omitempty"`
 
-	// Allow applying a subject transform to incoming messages before doing anything else
-	SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`
-
 	// Allow republish of the message after being sequenced and stored.
 	RePublish *RePublish `json:"republish,omitempty"`
 
@@ -85,12 +82,6 @@ type StreamConfig struct {
 	AllowRollup bool `json:"allow_rollup_hdrs"`
 }
 
-// SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received
-type SubjectTransformConfig struct {
-	Source      string `json:"src,omitempty"`
-	Destination string `json:"dest"`
-}
-
 // RePublish is for republishing messages once committed to a stream.
 type RePublish struct {
 	Source      string `json:"src,omitempty"`
@@ -163,23 +154,20 @@ type PeerInfo struct {
 
 // StreamSourceInfo shows information about an upstream stream source.
 type StreamSourceInfo struct {
-	Name                 string          `json:"name"`
-	External             *ExternalStream `json:"external,omitempty"`
-	Lag                  uint64          `json:"lag"`
-	Active               time.Duration   `json:"active"`
-	Error                *ApiError       `json:"error,omitempty"`
-	FilterSubject        string          `json:"filter_subject,omitempty"`
-	SubjectTransformDest string          `json:"subject_transform_dest,omitempty"`
+	Name     string          `json:"name"`
+	External *ExternalStream `json:"external,omitempty"`
+	Lag      uint64          `json:"lag"`
+	Active   time.Duration   `json:"active"`
+	Error    *ApiError       `json:"error,omitempty"`
 }
 
 // StreamSource dictates how streams can source from other streams.
 type StreamSource struct {
-	Name                 string          `json:"name"`
-	OptStartSeq          uint64          `json:"opt_start_seq,omitempty"`
-	OptStartTime         *time.Time      `json:"opt_start_time,omitempty"`
-	FilterSubject        string          `json:"filter_subject,omitempty"`
-	SubjectTransformDest string          `json:"subject_transform_dest,omitempty"`
-	External             *ExternalStream `json:"external,omitempty"`
+	Name          string          `json:"name"`
+	OptStartSeq   uint64          `json:"opt_start_seq,omitempty"`
+	OptStartTime  *time.Time      `json:"opt_start_time,omitempty"`
+	FilterSubject string          `json:"filter_subject,omitempty"`
+	External      *ExternalStream `json:"external,omitempty"`
 
 	// Internal
 	iname string // For indexing when stream names are the same for multiple sources.
@@ -232,11 +220,8 @@ type stream struct {
 	// Indicates we have direct consumers.
 	directs int
 
-	// For input subject transform
-	itr *subjectTransform
-
 	// For republishing.
-	tr *subjectTransform
+	tr *transform
 
 	// For processing consumers without main stream lock.
 	clsMu sync.RWMutex
@@ -286,8 +271,6 @@ type sourceInfo struct {
 	qch   chan struct{}
 	sip   bool // setup in progress
 	wg    sync.WaitGroup
-	sf    string // subject filter
-	tr    *subjectTransform
 }
 
 // For mirrors and direct get
@@ -421,17 +404,10 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		}
 	}
 
-	// Setup our internal indexed names here for sources and check if the transform (if any) is valid.
+	// Setup our internal indexed names here for sources.
 	if len(cfg.Sources) > 0 {
-		for i, ssi := range cfg.Sources {
-			ssi.setIndexName(i)
-			// check the transform, if any, is valid
-			if ssi.SubjectTransformDest != _EMPTY_ {
-				if _, err = newSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest); err != nil {
-					jsa.mu.Unlock()
-					return nil, fmt.Errorf("subject transform for the source not valid %w", err)
-				}
-			}
+		for _, ssi := range cfg.Sources {
+			ssi.setIndexName()
 		}
 	}
 
@@ -478,22 +454,16 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		mset.ackq = s.newIPQueue(qpfx + "acks") // of uint64
 	}
 
-	// Check for input subject transform
-	if cfg.SubjectTransform != nil {
-		tr, err := newSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
-		if err != nil {
-			jsa.mu.Unlock()
-			return nil, fmt.Errorf("stream input subject transform not valid %w", err)
-		}
-		mset.itr = tr
-	}
-
 	// Check for RePublish.
 	if cfg.RePublish != nil {
-		tr, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
+		// Empty same as all.
+		if cfg.RePublish.Source == _EMPTY_ {
+			cfg.RePublish.Source = fwcs
+		}
+		tr, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
-			return nil, fmt.Errorf("stream republish transform not valid %w", err)
+			return nil, fmt.Errorf("stream configuration for republish not valid")
 		}
 		// Assign our transform for republishing.
 		mset.tr = tr
@@ -593,14 +563,13 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	return mset, nil
 }
 
-// Sets the index name. Usually just the stream name and the source's index (so you can source from the same stream more than once)
-// but when the stream is external we will
+// Sets the index name. Usually just the stream name but when the stream is external we will
 // use additional information in case the stream names are the same.
-func (ssi *StreamSource) setIndexName(i int) {
+func (ssi *StreamSource) setIndexName() {
 	if ssi.External != nil {
-		ssi.iname = strconv.Itoa(i) + ":" + ssi.Name + ":" + getHash(ssi.External.ApiPrefix)
+		ssi.iname = ssi.Name + ":" + getHash(ssi.External.ApiPrefix)
 	} else {
-		ssi.iname = strconv.Itoa(i) + ":" + ssi.Name
+		ssi.iname = ssi.Name
 	}
 }
 
@@ -1075,10 +1044,28 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		return exists, cfg.MaxMsgSize, cfg.Subjects
 	}
 
+	hasFilterSubjectOverlap := func(filter string, streamSubs []string) bool {
+		if filter == _EMPTY_ || len(streamSubs) == 0 {
+			return true
+		}
+		for _, sub := range streamSubs {
+			if SubjectsCollide(sub, filter) {
+				return true
+			}
+		}
+		return false
+	}
+
 	var streamSubs []string
 	var deliveryPrefixes []string
 	var apiPrefixes []string
 
+	// Some errors we want to suppress on recovery.
+	var isRecovering bool
+	if js := s.getJetStream(); js != nil {
+		isRecovering = js.isMetaRecovering()
+	}
+
 	// Do some pre-checking for mirror config to avoid cycles in clustered mode.
 	if cfg.Mirror != nil {
 		if len(cfg.Subjects) > 0 {
@@ -1097,6 +1084,11 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 			if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
 				return StreamConfig{}, NewJSMirrorMaxMessageSizeTooBigError()
 			}
+			if !isRecovering && !hasFilterSubjectOverlap(cfg.Mirror.FilterSubject, subs) {
+				return StreamConfig{}, NewJSStreamInvalidConfigError(
+					fmt.Errorf("mirror '%s' filter subject '%s' does not overlap with any origin stream subject",
+						cfg.Mirror.Name, cfg.Mirror.FilterSubject))
+			}
 		}
 		if cfg.Mirror.External != nil {
 			if cfg.Mirror.External.DeliverPrefix != _EMPTY_ {
@@ -1132,6 +1124,11 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 				if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
 					return StreamConfig{}, NewJSSourceMaxMessageSizeTooBigError()
 				}
+				if !isRecovering && !hasFilterSubjectOverlap(src.FilterSubject, streamSubs) {
+					return StreamConfig{}, NewJSStreamInvalidConfigError(
+						fmt.Errorf("source '%s' filter subject '%s' does not overlap with any origin stream subject",
+							src.Name, src.FilterSubject))
+				}
 			}
 			if src.External == nil {
 				continue
@@ -1280,7 +1277,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		if formsCycle {
 			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish destination forms a cycle"))
 		}
-		if _, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
+		if _, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
 			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish not valid"))
 		}
 	}
@@ -1500,22 +1497,14 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 			for _, s := range ocfg.Sources {
 				current[s.iname] = s.FilterSubject
 			}
-			for i, s := range cfg.Sources {
-				s.setIndexName(i)
+			for _, s := range cfg.Sources {
+				s.setIndexName()
 				if oFilter, ok := current[s.iname]; !ok {
 					if mset.sources == nil {
 						mset.sources = make(map[string]*sourceInfo)
 					}
 					mset.cfg.Sources = append(mset.cfg.Sources, s)
-					si := &sourceInfo{name: s.Name, iname: s.iname, sf: s.FilterSubject}
-					// Check for transform.
-					if s.SubjectTransformDest != _EMPTY_ {
-						var err error
-						if si.tr, err = newSubjectTransform(s.FilterSubject, s.SubjectTransformDest); err != nil {
-							return err
-						}
-					}
-
+					si := &sourceInfo{name: s.Name, iname: s.iname}
 					mset.sources[s.iname] = si
 					mset.setStartingSequenceForSource(s.iname)
 					mset.setSourceConsumer(s.iname, si.sseq+1, time.Time{})
@@ -1568,7 +1557,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 		if cfg.RePublish.Source == _EMPTY_ {
 			cfg.RePublish.Source = fwcs
 		}
-		tr, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
+		tr, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
 			return fmt.Errorf("stream configuration for republish not valid")
@@ -1718,16 +1707,138 @@ func (mset *stream) sourcesInfo() (sis []*StreamSourceInfo) {
 	return sis
 }
 
+func gatherSourceMirrorSubjects(subjects []string, cfg *StreamConfig, acc *Account) ([]string, bool) {
+	var hasExt bool
+	var seen map[string]bool
+
+	if cfg.Mirror != nil {
+		subjs, localHasExt := acc.streamSourceSubjects(cfg.Mirror, make(map[string]bool))
+		if len(subjs) > 0 {
+			subjects = append(subjects, subjs...)
+		}
+		if localHasExt {
+			hasExt = true
+		}
+	} else if len(cfg.Sources) > 0 {
+		seen = make(map[string]bool)
+		for _, si := range cfg.Sources {
+			subjs, localHasExt := acc.streamSourceSubjects(si, seen)
+			if len(subjs) > 0 {
+				subjects = append(subjects, subjs...)
+			}
+			if localHasExt {
+				hasExt = true
+			}
+		}
+	}
+
+	return subjects, hasExt
+}
+
+// Return the subjects for a stream source.
+func (a *Account) streamSourceSubjects(ss *StreamSource, seen map[string]bool) (subjects []string, hasExt bool) {
+	if ss != nil && ss.External != nil {
+		return nil, true
+	}
+
+	s, js, _ := a.getJetStreamFromAccount()
+
+	if !s.JetStreamIsClustered() {
+		return a.streamSourceSubjectsNotClustered(ss.Name, seen)
+	} else {
+		return js.streamSourceSubjectsClustered(a.Name, ss.Name, seen)
+	}
+}
+
+func (js *jetStream) streamSourceSubjectsClustered(accountName, streamName string, seen map[string]bool) (subjects []string, hasExt bool) {
+	if seen[streamName] {
+		return nil, false
+	}
+
+	// We are clustered here so need to work through stream assignments.
+	sa := js.streamAssignment(accountName, streamName)
+	if sa == nil {
+		return nil, false
+	}
+	seen[streamName] = true
+
+	js.mu.RLock()
+	cfg := sa.Config
+	if len(cfg.Subjects) > 0 {
+		subjects = append(subjects, cfg.Subjects...)
+	}
+
+	// Check if we need to keep going.
+	var sources []*StreamSource
+	if cfg.Mirror != nil {
+		sources = append(sources, cfg.Mirror)
+	} else if len(cfg.Sources) > 0 {
+		sources = append(sources, cfg.Sources...)
+	}
+	js.mu.RUnlock()
+
+	if len(sources) > 0 {
+		var subjs []string
+		if acc, err := js.srv.lookupAccount(accountName); err == nil {
+			for _, ss := range sources {
+				subjs, hasExt = acc.streamSourceSubjects(ss, seen)
+				if len(subjs) > 0 {
+					subjects = append(subjects, subjs...)
+				}
+				if hasExt {
+					break
+				}
+			}
+		}
+	}
+
+	return subjects, hasExt
+}
+
+func (a *Account) streamSourceSubjectsNotClustered(streamName string, seen map[string]bool) (subjects []string, hasExt bool) {
+	if seen[streamName] {
+		return nil, false
+	}
+
+	mset, err := a.lookupStream(streamName)
+	if err != nil {
+		return nil, false
+	}
+	seen[streamName] = true
+
+	cfg := mset.config()
+	if len(cfg.Subjects) > 0 {
+		subjects = append(subjects, cfg.Subjects...)
+	}
+
+	var subjs []string
+	if cfg.Mirror != nil {
+		subjs, hasExt = a.streamSourceSubjects(cfg.Mirror, seen)
+		if len(subjs) > 0 {
+			subjects = append(subjects, subjs...)
+		}
+	} else if len(cfg.Sources) > 0 {
+		for _, si := range cfg.Sources {
+			subjs, hasExt = a.streamSourceSubjects(si, seen)
+			if len(subjs) > 0 {
+				subjects = append(subjects, subjs...)
+			}
+			if hasExt {
+				break
+			}
+		}
+	}
+
+	return subjects, hasExt
+}
+
 // Lock should be held
 func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
 	if si == nil {
 		return nil
 	}
 
-	var ssi = StreamSourceInfo{Name: si.name, Lag: si.lag, Error: si.err, FilterSubject: si.sf}
-	if si.tr != nil {
-		ssi.SubjectTransformDest = si.tr.dest
-	}
+	ssi := &StreamSourceInfo{Name: si.name, Lag: si.lag, Error: si.err}
 	// If we have not heard from the source, set Active to -1.
 	if si.last.IsZero() {
 		ssi.Active = -1
@@ -1747,7 +1858,7 @@ func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
 			DeliverPrefix: ext.DeliverPrefix,
 		}
 	}
-	return &ssi
+	return ssi
 }
 
 // Return our source info for our mirror.
@@ -2722,11 +2833,6 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 	// Hold onto the origin reply which has all the metadata.
 	hdr = genHeader(hdr, JSStreamSource, si.genSourceHeader(m.rply))
 
-	// Do the subject transform for the source if there's one
-	if si.tr != nil {
-		m.subj = si.tr.TransformSubject(m.subj)
-	}
-
 	var err error
 	// If we are clustered we need to propose this message to the underlying raft group.
 	if node != nil {
@@ -2763,10 +2869,7 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 // Generate a new style source header.
 func (si *sourceInfo) genSourceHeader(reply string) string {
 	var b strings.Builder
-	// strip the source index number from the iname
-	name := si.iname[strings.IndexRune(si.iname, ':')+1:]
-
-	b.WriteString(name)
+	b.WriteString(si.iname)
 	b.WriteByte(' ')
 	// Grab sequence as text here from reply subject.
 	var tsa [expectedNumReplyTokens]string
@@ -2861,16 +2964,11 @@ func (mset *stream) startingSequenceForSources() {
 	// Always reset here.
 	mset.sources = make(map[string]*sourceInfo)
 
-	for i, ssi := range mset.cfg.Sources {
+	for _, ssi := range mset.cfg.Sources {
 		if ssi.iname == _EMPTY_ {
-			ssi.setIndexName(i)
+			ssi.setIndexName()
 		}
-		si := &sourceInfo{name: ssi.Name, iname: ssi.iname, sf: ssi.FilterSubject}
-		// Check for transform.
-		if ssi.SubjectTransformDest != _EMPTY_ {
-			si.tr, _ = newSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest) // can not return an error because validated in AddStream
-		}
-
+		si := &sourceInfo{name: ssi.Name, iname: ssi.iname}
 		mset.sources[ssi.iname] = si
 	}
 
@@ -3583,15 +3681,6 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		return nil
 	}
 
-	// Apply the input subject transform if any
-	if mset.itr != nil {
-		ts, err := mset.itr.Match(subject)
-		if err == nil {
-			// no filtering: if the subject doesn't map the source of the transform, don't change it
-			subject = ts
-		}
-	}
-
 	var accName string
 	if mset.acc != nil {
 		accName = mset.acc.Name
diff --git a/server/subject_transform.go b/server/subject_transform.go
deleted file mode 100644
index 2dd553c4..00000000
--- a/server/subject_transform.go
+++ /dev/null
@@ -1,518 +0,0 @@
-// Copyright 2023 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.
-
-package server
-
-import (
-	"fmt"
-	"hash/fnv"
-	"regexp"
-	"strconv"
-	"strings"
-)
-
-var commaSeparatorRegEx = regexp.MustCompile(`,\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 subjectTransform function types
-const (
-	NoTransform int16 = iota
-	BadTransform
-	Partition
-	Wildcard
-	SplitFromLeft
-	SplitFromRight
-	SliceFromLeft
-	SliceFromRight
-	Split
-)
-
-// Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters.
-// 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 subjectTransform struct {
-	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
-}
-
-// SubjectTransform transforms subjects using mappings
-//
-// This API is not part of the public API and not subject to SemVer protections
-type SubjectTransform interface {
-	// TODO(dlc) - We could add in client here to allow for things like foo -> foo.$ACCOUNT
-	Match(string) (string, error)
-	TransformSubject(subject string) string
-	TransformTokenizedSubject(tokens []string) string
-}
-
-func newSubjectTransform(src, dest string) (*subjectTransform, error) {
-	// No source given is equivalent to the source being ">"
-	if src == _EMPTY_ {
-		src = fwcs
-	}
-
-	// Both entries need to be valid subjects.
-	sv, stokens, npwcs, hasFwc := subjectInfo(src)
-	dv, dtokens, dnpwcs, dHasFwc := subjectInfo(dest)
-
-	// Make sure both are valid, match fwc if present and there are no pwcs in the dest subject.
-	if !sv || !dv || dnpwcs > 0 || hasFwc != dHasFwc {
-		return nil, ErrBadSubject
-	}
-
-	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 {
-		// We need to count to make sure that the dest has token holders for the pwcs.
-		sti := make(map[int]int)
-		for i, token := range stokens {
-			if len(token) == 1 && token[0] == pwc {
-				sti[len(sti)+1] = i
-			}
-		}
-
-		nphs := 0
-		for _, token := range dtokens {
-			tranformType, transformArgWildcardIndexes, transfomArgInt, transformArgString, err := indexPlaceHolders(token)
-			if err != nil {
-				return nil, err
-			}
-
-			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 _, wildcardIndex := range transformArgWildcardIndexes {
-					if wildcardIndex > npwcs {
-						return nil, &mappingDestinationErr{fmt.Sprintf("%s: [%d]", token, wildcardIndex), ErrorMappingDestinationFunctionWildcardIndexOutOfRange}
-					}
-					stis = append(stis, sti[wildcardIndex])
-				}
-				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, tranformType)
-				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, stis)
-				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, transfomArgInt)
-				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, transformArgString)
-
-			}
-		}
-		if nphs < npwcs {
-			// not all wildcards are being used in the destination
-			return nil, &mappingDestinationErr{dest, ErrMappingDestinationNotUsingAllWildcards}
-		}
-	}
-
-	return &subjectTransform{
-		src:                  src,
-		dest:                 dest,
-		dtoks:                dtokens,
-		stoks:                stokens,
-		dtokmftypes:          dtokMappingFunctionTypes,
-		dtokmftokindexesargs: dtokMappingFunctionTokenIndexes,
-		dtokmfintargs:        dtokMappingFunctionIntArgs,
-		dtokmfstringargs:     dtokMappingFunctionStringArgs,
-	}, nil
-}
-
-func getMappingFunctionArgs(functionRegEx *regexp.Regexp, token string) []string {
-	commandStrings := functionRegEx.FindStringSubmatch(token)
-	if len(commandStrings) > 1 {
-		return commaSeparatorRegEx.Split(commandStrings[1], -1)
-	}
-	return nil
-}
-
-// 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 subjectTransform 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 {
-				// other things rely on tokens starting with $ so not an error just leave it as is
-				return NoTransform, []int{-1}, -1, _EMPTY_, nil
-			}
-			return Wildcard, []int{tp}, -1, _EMPTY_, nil
-		}
-
-		// New 'mustache' style mapping
-		if length > 4 && token[0] == '{' && token[1] == '{' && token[length-2] == '}' && token[length-1] == '}' {
-			// wildcard(wildcard token index) (equivalent to $)
-			args := getMappingFunctionArgs(wildcardMappingFunctionRegEx, token)
-			if args != nil {
-				if len(args) == 1 && args[0] == _EMPTY_ {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
-				}
-				if len(args) == 1 {
-					tokenIndex, err := strconv.Atoi(strings.Trim(args[0], " "))
-					if err != nil {
-						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-					}
-					return Wildcard, []int{tokenIndex}, -1, _EMPTY_, nil
-				} else {
-					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 BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
-				}
-				if len(args) >= 2 {
-					mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[0], " "))
-					if err != nil {
-						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-					}
-					var numPositions = len(args[1:])
-					tokenIndexes := make([]int, numPositions)
-					for ti, t := range args[1:] {
-						i, err := strconv.Atoi(strings.Trim(t, " "))
-						if err != nil {
-							return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-						}
-						tokenIndexes[ti] = i
-					}
-
-					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 BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrUnknownMappingDestinationFunction}
-		}
-	}
-	return NoTransform, []int{-1}, -1, _EMPTY_, nil
-}
-
-// Helper function to tokenize subjects with partial wildcards into formal transform destinations.
-// e.g. "foo.*.*" -> "foo.$1.$2"
-func transformTokenize(subject string) string {
-	// We need to make the appropriate markers for the wildcards etc.
-	i := 1
-	var nda []string
-	for _, token := range strings.Split(subject, tsep) {
-		if token == pwcs {
-			nda = append(nda, fmt.Sprintf("$%d", i))
-			i++
-		} else {
-			nda = append(nda, token)
-		}
-	}
-	return strings.Join(nda, tsep)
-}
-
-// Helper function to go from transform destination to a subject with partial wildcards and ordered list of placeholders
-// E.g.:
-//
-//	"bar" -> "bar", []
-//	"foo.$2.$1" -> "foo.*.*", ["$2","$1"]
-func transformUntokenize(subject string) (string, []string) {
-	var phs []string
-	var nda []string
-
-	for _, token := range strings.Split(subject, tsep) {
-		if len(token) > 1 && token[0] == '$' && token[1] >= '1' && token[1] <= '9' {
-			phs = append(phs, token)
-			nda = append(nda, pwcs)
-		} else {
-			nda = append(nda, token)
-		}
-	}
-	return strings.Join(nda, tsep), phs
-}
-
-func tokenizeSubject(subject string) []string {
-	// Tokenize the subject.
-	tsa := [32]string{}
-	tts := tsa[:0]
-	start := 0
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			tts = append(tts, subject[start:i])
-			start = i + 1
-		}
-	}
-	tts = append(tts, subject[start:])
-	return tts
-}
-
-// Match will take a literal published subject that is associated with a client and will match and subjectTransform
-// the subject if possible.
-//
-// This API is not part of the public API and not subject to SemVer protections
-func (tr *subjectTransform) Match(subject string) (string, error) {
-	// Special case: matches any and no no-op subjectTransform. May not be legal config for some features
-	// but specific validations made at subjectTransform create time
-	if (tr.src == fwcs || tr.src == _EMPTY_) && (tr.dest == fwcs || tr.dest == _EMPTY_) {
-		return subject, nil
-	}
-
-	tts := tokenizeSubject(subject)
-	if !isValidLiteralSubject(tts) {
-		return _EMPTY_, ErrBadSubject
-	}
-
-	if (tr.src == _EMPTY_ || tr.src == fwcs) || isSubsetMatch(tts, tr.src) {
-		return tr.TransformTokenizedSubject(tts), nil
-	}
-	return _EMPTY_, ErrNoTransforms
-}
-
-// TransformSubject transforms a subject
-//
-// This API is not part of the public API and not subject to SemVer protection
-func (tr *subjectTransform) TransformSubject(subject string) string {
-	return tr.TransformTokenizedSubject(tokenizeSubject(subject))
-}
-
-func (tr *subjectTransform) getHashPartition(key []byte, numBuckets int) string {
-	h := fnv.New32a()
-	_, _ = h.Write(key)
-
-	return strconv.Itoa(int(h.Sum32() % uint32(numBuckets)))
-}
-
-// Do a subjectTransform on the subject to the dest subject.
-func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string {
-	if len(tr.dtokmftypes) == 0 {
-		return tr.dest
-	}
-
-	var b strings.Builder
-
-	// We need to walk destination tokens and create the mapped subject pulling tokens or mapping functions
-	li := len(tr.dtokmftypes) - 1
-	for i, mfType := range tr.dtokmftypes {
-		if mfType == NoTransform {
-			// Break if fwc
-			if len(tr.dtoks[i]) == 1 && tr.dtoks[i][0] == fwc {
-				break
-			}
-			b.WriteString(tr.dtoks[i])
-		} else {
-			switch mfType {
-			case Partition:
-				var (
-					_buffer       [64]byte
-					keyForHashing = _buffer[:0]
-				)
-				for _, sourceToken := range tr.dtokmftokindexesargs[i] {
-					keyForHashing = append(keyForHashing, []byte(tokens[sourceToken])...)
-				}
-				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 && split == _EMPTY_) {
-						b.WriteString(tsep)
-					}
-				}
-			}
-		}
-
-		if i < li {
-			b.WriteByte(btsep)
-		}
-	}
-
-	// We may have more source tokens available. This happens with ">".
-	if tr.dtoks[len(tr.dtoks)-1] == fwcs {
-		for sli, i := len(tokens)-1, len(tr.stoks)-1; i < len(tokens); i++ {
-			b.WriteString(tokens[i])
-			if i < sli {
-				b.WriteByte(btsep)
-			}
-		}
-	}
-	return b.String()
-}
-
-// Reverse a subjectTransform.
-func (tr *subjectTransform) reverse() *subjectTransform {
-	if len(tr.dtokmftokindexesargs) == 0 {
-		rtr, _ := newSubjectTransform(tr.dest, tr.src)
-		return rtr
-	}
-	// If we are here we need to dynamically get the correct reverse
-	// of this subjectTransform.
-	nsrc, phs := transformUntokenize(tr.dest)
-	var nda []string
-	for _, token := range tr.stoks {
-		if token == pwcs {
-			if len(phs) == 0 {
-				// TODO(dlc) - Should not happen
-				return nil
-			}
-			nda = append(nda, phs[0])
-			phs = phs[1:]
-		} else {
-			nda = append(nda, token)
-		}
-	}
-	ndest := strings.Join(nda, tsep)
-	rtr, _ := newSubjectTransform(nsrc, ndest)
-	return rtr
-}


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 4d3f7836..0142344b 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -16,8 +16,10 @@ package server
 import (
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
+	"reflect"
 	"strconv"
 	"strings"
 	"sync"
@@ -46,6 +48,60 @@ func simpleAccountServer(t *testing.T) (*Server, *Account, *Account) {
 	return s, f, b
 }
 
+func TestPlaceHolderIndex(t *testing.T) {
+	testString := "$1"
+	transformType, indexes, nbPartitions, _, err := indexPlaceHolders(testString)
+	var position int32
+
+	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)}}"
+
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
+
+	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) }}"
+
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{wildcard(2)}}"
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	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 = "{{SplitFromRight(3,2)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != SplitFromRight || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{SliceFromLeft(2,2)}}"
+	transformType, indexes, sliceSize, _, err := indexPlaceHolders(testString)
+
+	if err != nil || transformType != SliceFromLeft || len(indexes) != 1 || indexes[0] != 2 || sliceSize != 2 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+}
+
 func TestRegisterDuplicateAccounts(t *testing.T) {
 	s, _, _ := simpleAccountServer(t)
 	if _, err := s.RegisterAccount("$foo"); err == nil {
@@ -3208,6 +3264,78 @@ func TestSamplingHeader(t *testing.T) {
 	test(false, http.Header{"traceparent": []string{"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00"}})
 }
 
+func TestSubjectTransforms(t *testing.T) {
+	shouldErr := func(src, dest string) {
+		t.Helper()
+		if _, err := newTransform(src, dest); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) {
+			t.Fatalf("Did not get an error for src=%q and dest=%q", src, dest)
+		}
+	}
+
+	shouldErr("foo.*.*", "bar.$2") // Must place all pwcs.
+
+	// Must be valid subjects.
+	shouldErr("foo", "")
+	shouldErr("foo..", "bar")
+
+	// Wildcards are allowed in src, but must be matched by token placements on the other side.
+	// e.g. foo.* -> bar.$1.
+	// Need to have as many pwcs as placements on other side.
+	shouldErr("foo.*", "bar.*")
+	shouldErr("foo.*", "bar.$2")                   // Bad pwc token identifier
+	shouldErr("foo.*", "bar.$1.>")                 // fwcs have to match.
+	shouldErr("foo.>", "bar.baz")                  // fwcs have to match.
+	shouldErr("foo.*.*", "bar.$2")                 // Must place all pwcs.
+	shouldErr("foo.*", "foo.$foo")                 // invalid $ value
+	shouldErr("foo.*", "foo.{{wildcard(2)}}")      // Mapping function being passed an out of range wildcard index
+	shouldErr("foo.*", "foo.{{unimplemented(1)}}") // Mapping trying to use an unknown mapping function
+	shouldErr("foo.*", "foo.{{partition(10)}}")    // Not enough arguments passed to the mapping function
+	shouldErr("foo.*", "foo.{{wildcard(foo)}}")    // Invalid argument passed to the mapping function
+	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()
+		tr, err := newTransform(src, dest)
+		if err != nil {
+			t.Fatalf("Got an error %v for src=%q and dest=%q", err, src, dest)
+		}
+		return tr
+	}
+
+	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()
+		tr := shouldBeOK(src, dest)
+		s, err := tr.Match(sample)
+		if err != nil {
+			t.Fatalf("Got an error %v when expecting a match for %q to %q", err, sample, expected)
+		}
+		if s != expected {
+			t.Fatalf("Dest does not match what was expected. Got %q, expected %q", s, expected)
+		}
+	}
+
+	shouldMatch("foo", "bar", "foo", "bar")
+	shouldMatch("foo.*.bar.*.baz", "req.$2.$1", "foo.A.bar.B.baz", "req.B.A")
+	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(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) {
 	conf := createConfFile(t, []byte(`
 		listen: 127.0.0.1:-1
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index e93116ce..ad718dc7 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -5544,6 +5544,39 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterSourceFilterSubjectUpdateFail(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "MSR", 3)
+	defer c.shutdown()
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	// Origin
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 2,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "S",
+		Sources:  []*nats.StreamSource{{Name: "TEST", FilterSubject: "notthere"}},
+		Replicas: 2,
+	})
+	require_Error(t, err)
+	require_Equal(t, err.Error(), "nats: source 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "M",
+		Mirror:   &nats.StreamSource{Name: "TEST", FilterSubject: "notthere"},
+		Replicas: 2,
+	})
+	require_Error(t, err)
+	require_Equal(t, err.Error(), "nats: mirror 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
+}
+
 func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
 	c := createJetStreamClusterWithTemplate(t, jsClusterMirrorSourceImportsTempl, "MS5", 5)
 	defer c.shutdown()
@@ -5592,8 +5625,9 @@ func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
 	createConsumer("M", "foo")
 	createConsumer("M", "bar")
 	createConsumer("M", "baz.foo")
-	expectFail("M", ".")
-	expectFail("M", ">.foo")
+	expectFail("M", "baz")
+	expectFail("M", "baz.1.2")
+	expectFail("M", "apple")
 
 	// Make sure wider scoped subjects work as well.
 	createConsumer("M", "*")
@@ -5618,6 +5652,9 @@ func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
 
 	createConsumer("S", "foo.1")
 	createConsumer("S", "bar.1")
+	expectFail("S", "baz")
+	expectFail("S", "baz.1")
+	expectFail("S", "apple")
 
 	// Now cross account stuff.
 	nc2, js2 := jsClientConnect(t, s, nats.UserInfo("rip", "pass"))
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 3463a32b..838e45be 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -530,11 +530,15 @@ func TestJetStreamConsumerWithStartTime(t *testing.T) {
 				OptStartTime:  &startTime,
 				AckPolicy:     AckExplicit,
 			})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			defer o.delete()
 
 			msg, err := nc.Request(o.requestNextMsgSubject(), nil, time.Second)
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			sseq, dseq, _, _, _ := replyInfo(msg.Reply)
 			if dseq != 1 {
 				t.Fatalf("Expected delivered seq of 1, got %d", dseq)
@@ -580,9 +584,13 @@ func TestJetStreamConsumerWithMultipleStartOptions(t *testing.T) {
 				},
 			}
 			req, err := json.Marshal(obsReq)
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			_, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, subj), req, time.Second)
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			nc.Close()
 			s.Shutdown()
 		})
@@ -627,7 +635,9 @@ func TestJetStreamConsumerMaxDeliveries(t *testing.T) {
 				AckWait:        ackWait,
 				MaxDeliver:     maxDeliver,
 			})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			// Wait for redeliveries to pile up.
@@ -687,7 +697,9 @@ func TestJetStreamPullConsumerDelayedFirstPullWithReplayOriginal(t *testing.T) {
 				AckPolicy:    AckExplicit,
 				ReplayPolicy: ReplayOriginal,
 			})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			// Force delay here which triggers the bug.
@@ -735,7 +747,9 @@ func TestJetStreamConsumerAckFloorFill(t *testing.T) {
 				DeliverSubject: sub.Subject,
 				AckPolicy:      AckExplicit,
 			})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			var first *nats.Msg
@@ -791,7 +805,9 @@ func TestJetStreamNoPanicOnRaceBetweenShutdownAndConsumerDelete(t *testing.T) {
 					Durable:   fmt.Sprintf("d%d", i),
 					AckPolicy: AckExplicit,
 				})
-				require_NoError(t, err)
+				if err != nil {
+					t.Fatalf("Expected no error, got %v", err)
+				}
 				defer o.delete()
 				cons = append(cons, o)
 			}
@@ -854,7 +870,9 @@ func TestJetStreamAddStreamMaxMsgSize(t *testing.T) {
 
 			tooBig := []byte("1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ")
 			resp, err := nc.Request("foo", tooBig, time.Second)
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error.Description != "message size exceeds maximum allowed" {
 				t.Fatalf("Expected to get an error for maximum message size, got %q", pa.Error)
 			}
@@ -895,7 +913,9 @@ func TestJetStreamAddStreamBadSubjects(t *testing.T) {
 	expectAPIErr := func(cfg StreamConfig) {
 		t.Helper()
 		req, err := json.Marshal(cfg)
-		require_NoError(t, err)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		resp, _ := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
 		var scResp JSApiStreamCreateResponse
 		if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -930,7 +950,9 @@ func TestJetStreamMaxConsumers(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	si, err := js.StreamInfo("MAXC")
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if si.Config.MaxConsumers != 1 {
 		t.Fatalf("Expected max of 1, got %d", si.Config.MaxConsumers)
 	}
@@ -1740,7 +1762,9 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			jreq, _ = json.Marshal(req)
 
 			resp, err := nc.Request(getSubj, jreq, 100*time.Millisecond)
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			if status := resp.Header.Get("Status"); !strings.HasPrefix(status, "404") {
 				t.Fatalf("Expected status code of 404")
 			}
@@ -1845,7 +1869,9 @@ func TestJetStreamSubjectFiltering(t *testing.T) {
 			// Now let's check the messages
 			for i := 1; i <= toSend; i++ {
 				m, err := sub.NextMsg(time.Second)
-				require_NoError(t, err)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
 				// JetStream will have the subject match the stream subject, not delivery subject.
 				// We want these to only be subjB.
 				if m.Subject != subjB {
@@ -1915,7 +1941,9 @@ func TestJetStreamWorkQueueSubjectFiltering(t *testing.T) {
 			getNext := func(seqno int) {
 				t.Helper()
 				nextMsg, err := nc.Request(o.requestNextMsgSubject(), nil, time.Second)
-				require_NoError(t, err)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
 				if nextMsg.Subject != subjA {
 					t.Fatalf("Expected subject of %q, got %q", subjA, nextMsg.Subject)
 				}
@@ -2236,7 +2264,9 @@ func TestJetStreamWorkQueueRetentionStream(t *testing.T) {
 
 			// We will create a non-partitioned consumer. This should succeed.
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			// Now if we create another this should fail, only can have one non-partitioned.
@@ -2257,12 +2287,16 @@ func TestJetStreamWorkQueueRetentionStream(t *testing.T) {
 				return &ConsumerConfig{Durable: dname, FilterSubject: pname, AckPolicy: AckExplicit}
 			}
 			o, err = mset.addConsumer(pConfig("MY_WORK_QUEUE.A"))
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			// Now creating another with separate partition should work.
 			o2, err := mset.addConsumer(pConfig("MY_WORK_QUEUE.B"))
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o2.delete()
 
 			// Anything that would overlap should fail though.
@@ -2274,7 +2308,9 @@ func TestJetStreamWorkQueueRetentionStream(t *testing.T) {
 			}
 
 			o3, err := mset.addConsumer(pConfig("MY_WORK_QUEUE.C"))
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 
 			o.delete()
 			o2.delete()
@@ -2449,7 +2485,9 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			}
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			expectPending := func(ep int) {
@@ -2516,7 +2554,9 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			// Now do filtered consumers.
 			o.delete()
 			o, err = mset.addConsumer(&ConsumerConfig{Durable: "PBO-FILTERED", AckPolicy: AckExplicit, FilterSubject: "foo.22"})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			for i := 0; i < toSend; i++ {
@@ -2597,7 +2637,9 @@ func TestJetStreamAckReplyStreamPendingWithAcks(t *testing.T) {
 				FilterSubject:  "foo",
 				DeliverSubject: dsubj,
 			})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			if info := o.info(); int(info.NumPending) != toSend {
@@ -2659,7 +2701,9 @@ func TestJetStreamWorkQueueAckWaitRedelivery(t *testing.T) {
 			ackWait := 100 * time.Millisecond
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit, AckWait: ackWait})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			sub, _ := nc.SubscribeSync(nats.NewInbox())
@@ -2755,7 +2799,9 @@ func TestJetStreamWorkQueueNakRedelivery(t *testing.T) {
 			}
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			getMsg := func(sseq, dseq int) *nats.Msg {
@@ -2830,7 +2876,9 @@ func TestJetStreamWorkQueueWorkingIndicator(t *testing.T) {
 			ackWait := 100 * time.Millisecond
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit, AckWait: ackWait})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			getMsg := func(sseq, dseq int) *nats.Msg {
@@ -2911,7 +2959,9 @@ func TestJetStreamWorkQueueTerminateDelivery(t *testing.T) {
 			ackWait := 25 * time.Millisecond
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit, AckWait: ackWait})
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			getMsg := func(sseq, dseq int) *nats.Msg {
@@ -3291,7 +3341,9 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Data = []byte("HELLO")
 	m.Header.Set(JSExpectedStream, mname)
 	resp, err := nc.RequestMsg(m, 100*time.Millisecond)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error != nil {
 		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
 	}
@@ -3299,7 +3351,9 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	// Now test that we get an error back when expecting a different stream.
 	m.Header.Set(JSExpectedStream, "ORDERS")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3308,7 +3362,9 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Header.Set(JSExpectedStream, mname)
 	m.Header.Set(JSExpectedLastSeq, "10")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3316,7 +3372,9 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	// Or if we expect that there are no messages by setting "0" as the expected last seq
 	m.Header.Set(JSExpectedLastSeq, "0")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3333,7 +3391,9 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Header.Set(JSMsgId, "ZZZ")
 	m.Header.Set(JSExpectedLastMsgId, "BBB")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3356,7 +3416,9 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Header.Set(JSExpectedLastMsgId, "AAA")
 	m.Header.Set(JSMsgId, "BBB")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error != nil {
 		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
 	}
@@ -3407,7 +3469,9 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	sendStreamMsg(t, nc, mname, "Hello World!")
 
 	msg, err := nc.Request(rqn, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	_, dseq, dc, _, _ := replyInfo(msg.Reply)
 	if dseq != 1 {
 		t.Fatalf("Expected consumer sequence of 1, got %d", dseq)
@@ -3431,7 +3495,9 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	sendStreamMsg(t, nc, mname, "Hello World!")
 
 	msg, err = nc.Request(rqn, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	_, dseq, dc, _, _ = replyInfo(msg.Reply)
 	if dseq != 2 {
 		t.Fatalf("Expected consumer sequence of 2, got %d", dseq)
@@ -3484,7 +3550,9 @@ func TestJetStreamConsumerRateLimit(t *testing.T) {
 		DeliverSubject: "to",
 		RateLimit:      rateLimit,
 		AckPolicy:      AckNone})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer o.delete()
 
 	var received int
@@ -3830,7 +3898,9 @@ func TestJetStreamRedeliveryAfterServerRestart(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		AckWait:        100 * time.Millisecond,
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer o.delete()
 
 	checkFor(t, 250*time.Millisecond, 10*time.Millisecond, func() error {
@@ -3946,7 +4016,9 @@ func TestJetStreamSnapshots(t *testing.T) {
 	r.Reset(snapshot)
 
 	mset, err = acc.RestoreStream(&info.cfg, r)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	// Now compare to make sure they are equal.
 	if nusage := acc.JetStreamUsage(); !reflect.DeepEqual(nusage, pusage) {
 		t.Fatalf("Usage does not match after restore: %+v vs %+v", nusage, pusage)
@@ -3978,7 +4050,9 @@ func TestJetStreamSnapshots(t *testing.T) {
 	acc = s2.GlobalAccount()
 	r.Reset(snapshot)
 	mset, err = acc.RestoreStream(&info.cfg, r)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	o := mset.lookupConsumer("WQ-1")
 	if o == nil {
@@ -4052,7 +4126,9 @@ func TestJetStreamSnapshotsAPI(t *testing.T) {
 	}
 
 	o, err := mset.addConsumer(workerModeConfig("WQ"))
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	// Now grab some messages.
 	toReceive := rand.Intn(toSend) + 1
 	for r := 0; r < toReceive; r++ {
@@ -5336,7 +5412,9 @@ func TestJetStreamRedeliverAndLateAck(t *testing.T) {
 
 	nextSubj := o.requestNextMsgSubject()
 	msg, err := nc.Request(nextSubj, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	// Wait for past ackwait time
 	time.Sleep(150 * time.Millisecond)
@@ -7193,7 +7271,9 @@ func TestJetStreamPushConsumerFlowControl(t *testing.T) {
 	}
 
 	sub, err := nc.SubscribeSync(nats.NewInbox())
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	obsReq := CreateConsumerRequest{
@@ -7206,9 +7286,13 @@ func TestJetStreamPushConsumerFlowControl(t *testing.T) {
 		},
 	}
 	req, err := json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "dlc"), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7307,7 +7391,9 @@ func TestJetStreamPushConsumerIdleHeartbeats(t *testing.T) {
 	}
 
 	sub, err := nc.SubscribeSync(nats.NewInbox())
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	// Test errors first
@@ -7319,9 +7405,13 @@ func TestJetStreamPushConsumerIdleHeartbeats(t *testing.T) {
 		},
 	}
 	req, err := json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7332,9 +7422,13 @@ func TestJetStreamPushConsumerIdleHeartbeats(t *testing.T) {
 	// Set acceptable heartbeat.
 	obsReq.Config.Heartbeat = 100 * time.Millisecond
 	req, err = json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	ccResp.Error = nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7368,7 +7462,9 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithFilterSubject(t *testing.T) {
 
 	hbC := make(chan *nats.Msg, 8)
 	sub, err := nc.ChanSubscribe(nats.NewInbox(), hbC)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	obsReq := CreateConsumerRequest{
@@ -7381,9 +7477,13 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithFilterSubject(t *testing.T) {
 	}
 
 	req, err := json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7422,7 +7522,9 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithNoInterest(t *testing.T) {
 	dsubj := "d.22"
 	hbC := make(chan *nats.Msg, 8)
 	sub, err := nc.ChanSubscribe("d.>", hbC)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	obsReq := CreateConsumerRequest{
@@ -7434,9 +7536,13 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithNoInterest(t *testing.T) {
 	}
 
 	req, err := json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7474,7 +7580,9 @@ func TestJetStreamInfoAPIWithHeaders(t *testing.T) {
 	m.Data = []byte("HELLO-JS!")
 
 	resp, err := nc.RequestMsg(m, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	var info JSApiAccountInfoResponse
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
@@ -7509,7 +7617,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		MaxMsgs:  100,
 	}
 	req, err := json.Marshal(msetCfg)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, _ = nc.Request(fmt.Sprintf(JSApiStreamCreateT, msetCfg.Name), req, time.Second)
 	var scResp JSApiStreamCreateResponse
 	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -7534,7 +7644,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	msetCfg.Subjects = []string{"foo", "bar", "baz"}
 	msetCfg.MaxBytes = 2222222
 	req, err = json.Marshal(msetCfg)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, _ = nc.Request(fmt.Sprintf(JSApiStreamUpdateT, msetCfg.Name), req, time.Second)
 	scResp.Error, scResp.StreamInfo = nil, nil
 	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -7551,7 +7663,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Subjects: []string{"foo"},
 	}
 	req, err = json.Marshal(cfg)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, _ = nc.Request(fmt.Sprintf(JSApiStreamUpdateT, cfg.Name), req, time.Second)
 	scResp.Error, scResp.StreamInfo = nil, nil
 	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -7563,7 +7677,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Now lookup info again and see that we can see the new stream.
 	resp, err = nc.Request(JSApiAccountInfo, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if err = json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -7573,7 +7689,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Make sure list names works.
 	resp, err = nc.Request(JSApiStreams, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var namesResponse JSApiStreamNamesResponse
 	if err = json.Unmarshal(resp.Data, &namesResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7626,7 +7744,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	}
 
 	resp, err = nc.Request(fmt.Sprintf(JSApiStreamInfoT, msetCfg.Name), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var msi StreamInfo
 	if err = json.Unmarshal(resp.Data, &msi); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7640,7 +7760,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Looking up one that is not there should yield an error.
 	resp, err = nc.Request(fmt.Sprintf(JSApiStreamInfoT, "BOB"), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var bResp JSApiStreamInfoResponse
 	if err = json.Unmarshal(resp.Data, &bResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7654,9 +7776,13 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Config: ConsumerConfig{DeliverSubject: delivery},
 	}
 	req, err = json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, msetCfg.Name), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7682,7 +7808,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Check that we get an error if the stream name in the subject does not match the config.
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "BOB"), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	ccResp.Error, ccResp.ConsumerInfo = nil, nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7692,7 +7820,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Get the list of all of the consumers for our stream.
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumersT, msetCfg.Name), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var clResponse JSApiConsumerNamesResponse
 	if err = json.Unmarshal(resp.Data, &clResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7703,7 +7833,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	// Now let's get info about our consumer.
 	cName := clResponse.Consumers[0]
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerInfoT, msetCfg.Name, cName), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var oinfo ConsumerInfo
 	if err = json.Unmarshal(resp.Data, &oinfo); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7743,9 +7875,13 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Config: ConsumerConfig{Durable: "myd", DeliverSubject: delivery},
 	}
 	req, err = json.Marshal(obsReq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, msetCfg.Name), req, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	ccResp.Error, ccResp.ConsumerInfo = nil, nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7768,9 +7904,13 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Config: ConsumerConfig{DeliverSubject: delivery},
 	}
 	req2, err := json.Marshal(obsReq2)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, err = nc.Request(fmt.Sprintf(JSApiDurableCreateT, msetCfg.Name, obsReq.Config.Durable), req2, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	ccResp.Error, ccResp.ConsumerInfo = nil, nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7780,7 +7920,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	// Now delete a msg.
 	dreq := JSApiMsgDeleteRequest{Seq: 2}
 	dreqj, err := json.Marshal(dreq)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	resp, _ = nc.Request(fmt.Sprintf(JSApiMsgDeleteT, msetCfg.Name), dreqj, time.Second)
 	var delMsgResp JSApiMsgDeleteResponse
 	if err = json.Unmarshal(resp.Data, &delMsgResp); err != nil {
@@ -7816,7 +7958,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	// Now grab stats again.
 	// This will get the current information about usage and limits for this account.
 	resp, err = nc.Request(JSApiAccountInfo, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -7839,7 +7983,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		MaxStreams: 4,
 	}
 	req, err = json.Marshal(template)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	// Check that the name in config has to match the name in the subject
 	resp, _ = nc.Request(fmt.Sprintf(JSApiTemplateCreateT, "BOB"), req, time.Second)
@@ -7927,7 +8073,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	sendStreamMsg(t, nc, "kv.22", "derek")
 	// Last do info
 	resp, err = nc.Request(fmt.Sprintf(JSApiTemplateInfoT, "kv"), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var ti StreamTemplateInfo
 	if err = json.Unmarshal(resp.Data, &ti); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -8763,7 +8911,9 @@ func TestJetStreamTemplateFileStoreRecovery(t *testing.T) {
 		t.Fatalf("Expected 100 auto-created streams, got %d", nms)
 	}
 	tmpl, err := acc.lookupStreamTemplate(template.Name)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	// Make sure t.delete() survives restart.
 	tmpl.delete()
 
@@ -8933,7 +9083,9 @@ func TestJetStreamMultipleAccountsBasics(t *testing.T) {
 	defer ncb.Close()
 
 	resp, err := ncb.Request(JSApiAccountInfo, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var info JSApiAccountInfoResponse
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -9022,7 +9174,9 @@ func TestJetStreamMultipleAccountsBasics(t *testing.T) {
 	// Now check that limits have been updated.
 	// Account B
 	resp, err = ncb.Request(JSApiAccountInfo, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -9042,7 +9196,9 @@ func TestJetStreamMultipleAccountsBasics(t *testing.T) {
 
 	// Account C
 	resp, err = ncc.Request(JSApiAccountInfo, nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -9112,7 +9268,9 @@ func TestJetStreamStoreDirectoryFix(t *testing.T) {
 	}
 	// Push based.
 	sub, err := js.SubscribeSync("TEST", nats.Durable("dlc"))
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	// Now shutdown the server.
@@ -9166,14 +9324,20 @@ func TestJetStreamPushConsumersPullError(t *testing.T) {
 	}
 	// Push based.
 	sub, err := js.SubscribeSync("TEST")
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 	ci, err := sub.ConsumerInfo()
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	// Now do a pull. Make sure we get an error.
 	m, err := nc.Request(fmt.Sprintf(JSApiRequestNextT, "TEST", ci.Name), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if m.Header.Get("Status") != "409" {
 		t.Fatalf("Expected a 409 status code, got %q", m.Header.Get("Status"))
 	}
@@ -9370,7 +9534,9 @@ func TestJetStreamPubWithSyncPerf(t *testing.T) {
 	defer nc.Close()
 
 	_, err := js.AddStream(&nats.StreamConfig{Name: "foo"})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	toSend := 1_000_000
 	payload := []byte("Hello World")
@@ -10078,8 +10244,9 @@ func TestJetStreamConsumerMaxAckPending(t *testing.T) {
 				AckPolicy:      AckExplicit,
 				MaxAckPending:  maxAckPending,
 			})
-			require_NoError(t, err)
-
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			sub, _ := nc.SubscribeSync(o.info().Config.DeliverSubject)
@@ -10121,8 +10288,9 @@ func TestJetStreamConsumerMaxAckPending(t *testing.T) {
 				AckPolicy:      AckExplicit,
 				MaxAckPending:  maxAckPending,
 			})
-			require_NoError(t, err)
-
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			sub, _ = nc.SubscribeSync(o.info().Config.DeliverSubject)
@@ -10188,8 +10356,9 @@ func TestJetStreamPullConsumerMaxAckPending(t *testing.T) {
 				AckPolicy:     AckExplicit,
 				MaxAckPending: maxAckPending,
 			})
-			require_NoError(t, err)
-
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			getSubj := o.requestNextMsgSubject()
@@ -10285,8 +10454,9 @@ func TestJetStreamPullConsumerMaxAckPendingRedeliveries(t *testing.T) {
 				AckWait:       ackWait,
 				MaxAckPending: maxAckPending,
 			})
-			require_NoError(t, err)
-
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
 			defer o.delete()
 
 			getSubj := o.requestNextMsgSubject()
@@ -10510,7 +10680,9 @@ func TestJetStreamAccountImportBasics(t *testing.T) {
 
 	// We mapped the next message request, "$JS.API.CONSUMER.MSG.NEXT.ORDERS.d" -> "nxt.msg"
 	m, err := nc.Request("nxt.msg", nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if string(m.Data) != "ORDERS-1" {
 		t.Fatalf("Expected to receive %q, got %q", "ORDERS-1", m.Data)
 	}
@@ -10612,7 +10784,9 @@ func TestJetStreamAccountImportAll(t *testing.T) {
 
 	// This will get the current information about usage and limits for this account.
 	resp, err := nc.Request(mapSubj(JSApiAccountInfo), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var info JSApiAccountInfoResponse
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -10622,7 +10796,9 @@ func TestJetStreamAccountImportAll(t *testing.T) {
 	}
 	// Lookup streams.
 	resp, err = nc.Request(mapSubj(JSApiStreams), nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var namesResponse JSApiStreamNamesResponse
 	if err = json.Unmarshal(resp.Data, &namesResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -10682,7 +10858,9 @@ func TestJetStreamServerReload(t *testing.T) {
 		t.Fatalf("Unexpected error looking up account: %v", err)
 	}
 	mset, err := acc.addStream(&StreamConfig{Name: "22"})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	toSend := 10
 	for i := 0; i < toSend; i++ {
@@ -10752,7 +10930,9 @@ func TestJetStreamConfigReloadWithGlobalAccount(t *testing.T) {
 		}
 	}
 	si, err := js.StreamInfo("foo")
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if si.State.Msgs != uint64(toSend) {
 		t.Fatalf("Expected %d msgs after restart, got %d", toSend, si.State.Msgs)
 	}
@@ -11088,7 +11268,9 @@ func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo", "bar", "baz", "N.*"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	// Add in some messages.
 	js.Publish("foo", []byte("OK"))
@@ -11100,8 +11282,9 @@ func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js.StreamInfo("TEST")
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 15 {
 			return fmt.Errorf("Expected 15 msgs, got state: %+v", si.State)
 		}
@@ -11110,8 +11293,9 @@ func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 
 	checkWider := func(subj string, numExpected int) {
 		sub, err := js.SubscribeSync(subj)
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		defer sub.Unsubscribe()
 		checkSubsPending(t, sub, numExpected)
 	}
@@ -11136,14 +11320,17 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo", "bar", "baz.*"},
 	})
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	// Create Mirror now.
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name:   "M",
 		Mirror: &nats.StreamSource{Name: "TEST"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	dsubj := nats.NewInbox()
 	nc.SubscribeSync(dsubj)
@@ -11152,13 +11339,24 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 	createConsumer := func(sn, fs string) {
 		t.Helper()
 		_, err = js.AddConsumer(sn, &nats.ConsumerConfig{DeliverSubject: dsubj, FilterSubject: fs})
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+	expectFail := func(sn, fs string) {
+		t.Helper()
+		_, err = js.AddConsumer(sn, &nats.ConsumerConfig{DeliverSubject: dsubj, FilterSubject: fs})
+		if err == nil {
+			t.Fatalf("Expected error but got none")
+		}
 	}
 
 	createConsumer("M", "foo")
 	createConsumer("M", "bar")
 	createConsumer("M", "baz.foo")
+	expectFail("M", "baz")
+	expectFail("M", "baz.1.2")
+	expectFail("M", "apple")
 
 	// Now do some sources.
 	if _, err := js.AddStream(&nats.StreamConfig{Name: "O1", Subjects: []string{"foo.*"}}); err != nil {
@@ -11173,10 +11371,15 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 		Name:    "S",
 		Sources: []*nats.StreamSource{{Name: "O1"}, {Name: "O2"}},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	createConsumer("S", "foo.1")
 	createConsumer("S", "bar.1")
+	expectFail("S", "baz")
+	expectFail("S", "baz.1")
+	expectFail("S", "apple")
 
 	// Chaining
 	// Create Mirror now.
@@ -11184,11 +11387,16 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 		Name:   "M2",
 		Mirror: &nats.StreamSource{Name: "M"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	createConsumer("M2", "foo")
 	createConsumer("M2", "bar")
 	createConsumer("M2", "baz.foo")
+	expectFail("M2", "baz")
+	expectFail("M2", "baz.1.2")
+	expectFail("M2", "apple")
 }
 
 func TestJetStreamMirrorBasics(t *testing.T) {
@@ -11248,12 +11456,14 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	// Faster timeout since we loop below checking for condition.
 	js2, err := nc.JetStream(nats.MaxWait(250 * time.Millisecond))
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M1")
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 100 {
 			return fmt.Errorf("Expected 100 msgs, got state: %+v", si.State)
 		}
@@ -11282,8 +11492,9 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M2")
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 50 {
 			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
 		}
@@ -11309,8 +11520,9 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M3")
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 101 {
 			return fmt.Errorf("Expected 101 msgs, got state: %+v", si.State)
 		}
@@ -11330,8 +11542,9 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M4")
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 150 {
 			return fmt.Errorf("Expected 150 msgs, got state: %+v", si.State)
 		}
@@ -11373,11 +11586,13 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	createStream := func(cfg *StreamConfig) {
 		t.Helper()
 		req, err := json.Marshal(cfg)
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		var resp JSApiStreamCreateResponse
 		if err := json.Unmarshal(rm.Data, &resp); err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -11408,7 +11623,7 @@ func TestJetStreamSourceBasics(t *testing.T) {
 		Name:    "MS",
 		Storage: FileStorage,
 		Sources: []*StreamSource{
-			{Name: "foo", SubjectTransformDest: "foo2.>"},
+			{Name: "foo"},
 			{Name: "bar"},
 			{Name: "baz"},
 		},
@@ -11418,25 +11633,20 @@ func TestJetStreamSourceBasics(t *testing.T) {
 
 	// Faster timeout since we loop below checking for condition.
 	js2, err := nc.JetStream(nats.MaxWait(250 * time.Millisecond))
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("MS")
-		require_NoError(t, err)
-
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 50 {
 			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
 		}
 		return nil
 	})
 
-	m, err := js.GetMsg("MS", 1)
-	require_NoError(t, err)
-
-	if m.Subject != "foo2.foo" {
-		t.Fatalf("Expected message subject foo2.foo, got %s", m.Subject)
-	}
-
 	// Test Source Updates
 	ncfg := &nats.StreamConfig{
 		Name: "MS",
@@ -11469,15 +11679,19 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	createStream(cfg)
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("FMS")
-		require_NoError(t, err)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 25 {
 			return fmt.Errorf("Expected 25 msgs, got state: %+v", si.State)
 		}
 		return nil
 	})
 	// Double check first starting.
-	m, err = js.GetMsg("FMS", 1)
-	require_NoError(t, err)
+	m, err := js.GetMsg("FMS", 1)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if shdr := m.Header.Get(JSStreamSource); shdr == _EMPTY_ {
 		t.Fatalf("Expected a header, got none")
 	} else if _, sseq := streamAndSeq(shdr); sseq != 26 {
@@ -11495,7 +11709,9 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	createStream(cfg)
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("FMS2")
-		require_NoError(t, err)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		if si.State.Msgs != 20 {
 			return fmt.Errorf("Expected 20 msgs, got state: %+v", si.State)
 		}
@@ -11513,46 +11729,6 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	}
 }
 
-func TestJetStreamInputTransform(t *testing.T) {
-	s := RunBasicJetStreamServer(t)
-	defer s.Shutdown()
-
-	// Client for API requests.
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	createStream := func(cfg *StreamConfig) {
-		t.Helper()
-		req, err := json.Marshal(cfg)
-		require_NoError(t, err)
-
-		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
-		require_NoError(t, err)
-
-		var resp JSApiStreamCreateResponse
-		if err := json.Unmarshal(rm.Data, &resp); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if resp.Error != nil {
-			t.Fatalf("Unexpected error: %+v", resp.Error)
-		}
-	}
-
-	createStream(&StreamConfig{Name: "T1", Subjects: []string{"foo"}, SubjectTransform: &SubjectTransformConfig{Source: ">", Destination: "transformed.>"}, Storage: MemoryStorage})
-
-	// publish a message
-	if _, err := js.Publish("foo", []byte("OK")); err != nil {
-		t.Fatalf("Unexpected publish error: %v", err)
-	}
-
-	m, err := js.GetMsg("T1", 1)
-	require_NoError(t, err)
-
-	if m.Subject != "transformed.foo" {
-		t.Fatalf("Expected message subject transformed.foo, got %s", m.Subject)
-	}
-}
-
 func TestJetStreamOperatorAccounts(t *testing.T) {
 	s, _ := RunServerWithConfig("./configs/js-op.conf")
 	if config := s.JetStreamConfig(); config != nil {
@@ -11583,7 +11759,9 @@ func TestJetStreamOperatorAccounts(t *testing.T) {
 	defer s.Shutdown()
 
 	jsz, err := s.Jsz(nil)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	if jsz.Streams != 1 {
 		t.Fatalf("Expected jsz to report our stream on restart")
@@ -11681,7 +11859,9 @@ func TestJetStreamDomainInPubAck(t *testing.T) {
 
 	// Check by hand for now til it makes its way into Go client.
 	am, err := nc.Request("foo", nil, time.Second)
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	var pa PubAck
 	json.Unmarshal(am.Data, &pa)
 	if pa.Domain != "HUB" {
@@ -11702,7 +11882,9 @@ func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name: "S",
@@ -11710,7 +11892,9 @@ func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 			Name: "TEST",
 		}},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	if _, err = js.Publish("foo", nil); err != nil {
 		t.Fatalf("Unexpected publish error: %v", err)
@@ -11727,7 +11911,9 @@ func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 	})
 
 	si, err := js.StreamInfo("TEST")
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	// Direct consumers should not be reported
 	if si.State.Consumers != 0 {
@@ -11766,7 +11952,9 @@ func TestJetStreamTemplatedErrorsBug(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	_, err = js.PullSubscribe("foo", "")
 	if err != nil && strings.Contains(err.Error(), "{err}") {
@@ -11882,8 +12070,9 @@ func TestJetStreamServerEncryption(t *testing.T) {
 				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"))
-				require_NoError(t, err)
-
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
 				if _, err := decodeConsumerState(state); err == nil {
 					t.Fatalf("Expected decoding consumer state to fail")
 				}
@@ -11903,8 +12092,9 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			defer nc.Close()
 
 			si2, err := js.StreamInfo("TEST")
-			require_NoError(t, err)
-
+			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)
 			}
@@ -11935,7 +12125,9 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			// Now test snapshots etc.
 			acc := s.GlobalAccount()
 			mset, err := acc.lookupStream("TEST")
-			require_NoError(t, err)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			scfg := mset.config()
 			sr, err := mset.snapshot(5*time.Second, false, true)
 			if err != nil {
@@ -11953,7 +12145,9 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			nacc := ns.GlobalAccount()
 			r := bytes.NewReader(snapshot)
 			mset, err = nacc.RestoreStream(&scfg, r)
-			require_NoError(t, err)
+			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)
@@ -11967,7 +12161,9 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			acc = s.GlobalAccount()
 			r.Reset(snapshot)
 			mset, err = acc.RestoreStream(&scfg, r)
-			require_NoError(t, err)
+			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)
@@ -11992,7 +12188,9 @@ func TestJetStreamConsumerBadNumPending(t *testing.T) {
 		Name:     "ORDERS",
 		Subjects: []string{"orders.*"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	newOrders := func(n int) {
 		// Queue up new orders.
@@ -12009,13 +12207,15 @@ func TestJetStreamConsumerBadNumPending(t *testing.T) {
 	}
 
 	op, err := js.Subscribe("orders.created", process)
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer op.Unsubscribe()
 
 	mon, err := js.SubscribeSync("orders.*")
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer mon.Unsubscribe()
 
 	waitForMsgs := func(n uint64) {
@@ -12250,10 +12450,14 @@ func TestJetStreamConsumerCleanupWithRetentionPolicy(t *testing.T) {
 		Subjects:  []string{"orders.*"},
 		Retention: nats.InterestPolicy,
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	sub, err := js.SubscribeSync("orders.*")
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	payload := []byte("Hello World")
 	for i := 0; i < 10; i++ {
@@ -12278,8 +12482,9 @@ func TestJetStreamConsumerCleanupWithRetentionPolicy(t *testing.T) {
 
 	acc := s.GlobalAccount()
 	mset, err := acc.lookupStream("ORDERS")
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	o := mset.lookupConsumer(ci.Name)
 	if o == nil {
 		t.Fatalf("Error looking up consumer %q", ci.Name)
@@ -12310,13 +12515,16 @@ func TestJetStreamPurgeEffectsConsumerDelivery(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo.*"},
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	js.Publish("foo.a", []byte("show once"))
 
 	sub, err := js.SubscribeSync("foo.*", nats.AckWait(250*time.Millisecond), nats.DeliverAll(), nats.AckExplicit())
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	checkSubsPending(t, sub, 1)
@@ -12358,11 +12566,14 @@ func TestJetStreamExpireCausesDeadlock(t *testing.T) {
 		MaxMsgs:   10,
 		Retention: nats.InterestPolicy,
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	sub, err := js.SubscribeSync("foo.bar")
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer sub.Unsubscribe()
 
 	// Publish from two connections to get the write lock request wedged in between
@@ -12463,8 +12674,9 @@ func TestJetStreamDefaultMaxMsgsPer(t *testing.T) {
 		Storage:  nats.MemoryStorage,
 		MaxMsgs:  10,
 	})
-	require_NoError(t, err)
-
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	if si.Config.MaxMsgsPerSubject != -1 {
 		t.Fatalf("Expected default of -1, got %d", si.Config.MaxMsgsPerSubject)
 	}
@@ -12484,7 +12696,9 @@ func TestJetStreamBadConsumerCreateErr(t *testing.T) {
 		Subjects: []string{"foo.*"},
 		Storage:  nats.MemoryStorage,
 	})
-	require_NoError(t, err)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 
 	// When adding a consumer with both deliver subject and max wait (push vs pull),
 	// we got the wrong err about deliver subject having a wildcard.
@@ -16594,25 +16808,39 @@ func TestJetStreamPullMaxBytes(t *testing.T) {
 	req = &JSApiConsumerGetNextRequest{Batch: 1, MaxBytes: 10_000_000, NoWait: true}
 	jreq, _ = json.Marshal(req)
 	nc.PublishRequest(subj, reply, jreq)
-	checkSubsPending(t, sub, 1)
+	// we expect two messages, as the second one should be `Batch Completed` status.
+	checkSubsPending(t, sub, 2)
 
+	// first one is message from the stream.
 	m, err = sub.NextMsg(time.Second)
 	require_NoError(t, err)
 	require_True(t, len(m.Data) == dsz)
 	require_True(t, len(m.Header) == 0)
+	// second one is the status.
+	m, err = sub.NextMsg(time.Second)
+	require_NoError(t, err)
+	if v := m.Header.Get("Description"); v != "Batch Completed" {
+		t.Fatalf("Expected Batch Completed, got: %s", v)
+	}
 	checkSubsPending(t, sub, 0)
 
 	// Same but with batch > 1
 	req = &JSApiConsumerGetNextRequest{Batch: 5, MaxBytes: 10_000_000, NoWait: true}
 	jreq, _ = json.Marshal(req)
 	nc.PublishRequest(subj, reply, jreq)
-	checkSubsPending(t, sub, 5)
+	// 6, not 5, as 6th is the status.
+	checkSubsPending(t, sub, 6)
 	for i := 0; i < 5; i++ {
 		m, err = sub.NextMsg(time.Second)
 		require_NoError(t, err)
 		require_True(t, len(m.Data) == dsz)
 		require_True(t, len(m.Header) == 0)
 	}
+	m, err = sub.NextMsg(time.Second)
+	require_NoError(t, err)
+	if v := m.Header.Get("Description"); v != "Batch Completed" {
+		t.Fatalf("Expected Batch Completed, got: %s", v)
+	}
 	checkSubsPending(t, sub, 0)
 
 	// Now ask for large batch but make sure we are limited by batch size.
@@ -19119,3 +19347,56 @@ func TestJetStreamMsgBlkFailOnKernelFault(t *testing.T) {
 			friendlyBytes(si.Config.MaxBytes), friendlyBytes(int64(si.State.Bytes)))
 	}
 }
+
+func TestJetStreamPullConsumerBatchCompleted(t *testing.T) {
+
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	msgSize := 128
+	msg := make([]byte, msgSize)
+	rand.Read(msg)
+
+	for i := 0; i < 10; i++ {
+		_, err := js.Publish("foo", msg)
+		require_NoError(t, err)
+	}
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "dur",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	req := JSApiConsumerGetNextRequest{Batch: 0, MaxBytes: 1024, Expires: 250 * time.Millisecond}
+
+	reqb, _ := json.Marshal(req)
+	sub := natsSubSync(t, nc, nats.NewInbox())
+	err = nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.dur", sub.Subject, reqb)
+	require_NoError(t, err)
+
+	// Expect first message to arrive normally.
+	_, err = sub.NextMsg(time.Second * 1)
+	require_NoError(t, err)
+
+	// Second message should be info that batch is complete, but there were pending bytes.
+	pullMsg, err := sub.NextMsg(time.Second * 1)
+	require_NoError(t, err)
+
+	if v := pullMsg.Header.Get("Status"); v != "409" {
+		t.Fatalf("Expected 409, got: %s", v)
+	}
+	if v := pullMsg.Header.Get("Description"); v != "Batch Completed" {
+		t.Fatalf("Expected Batch Completed, got: %s", v)
+	}
+
+}
diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go
deleted file mode 100644
index 8ac9b6ca..00000000
--- a/server/subject_transform_test.go
+++ /dev/null
@@ -1,148 +0,0 @@
-// Copyright 2023 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.
-
-package server
-
-import (
-	"errors"
-	"reflect"
-	"testing"
-)
-
-func TestPlaceHolderIndex(t *testing.T) {
-	testString := "$1"
-	transformType, indexes, nbPartitions, _, err := indexPlaceHolders(testString)
-	var position int32
-
-	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)}}"
-
-	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
-
-	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) }}"
-
-	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{wildcard(2)}}"
-	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	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 = "{{SplitFromRight(3,2)}}"
-	transformType, indexes, position, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != SplitFromRight || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{SliceFromLeft(2,2)}}"
-	transformType, indexes, sliceSize, _, err := indexPlaceHolders(testString)
-
-	if err != nil || transformType != SliceFromLeft || len(indexes) != 1 || indexes[0] != 2 || sliceSize != 2 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-}
-
-func TestSubjectTransforms(t *testing.T) {
-	shouldErr := func(src, dest string) {
-		t.Helper()
-		if _, err := newSubjectTransform(src, dest); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) {
-			t.Fatalf("Did not get an error for src=%q and dest=%q", src, dest)
-		}
-	}
-
-	shouldErr("foo.*.*", "bar.$2") // Must place all pwcs.
-
-	// Must be valid subjects.
-	shouldErr("foo", "")
-	shouldErr("foo..", "bar")
-
-	// Wildcards are allowed in src, but must be matched by token placements on the other side.
-	// e.g. foo.* -> bar.$1.
-	// Need to have as many pwcs as placements on other side.
-	shouldErr("foo.*", "bar.*")
-	shouldErr("foo.*", "bar.$2")                   // Bad pwc token identifier
-	shouldErr("foo.*", "bar.$1.>")                 // fwcs have to match.
-	shouldErr("foo.>", "bar.baz")                  // fwcs have to match.
-	shouldErr("foo.*.*", "bar.$2")                 // Must place all pwcs.
-	shouldErr("foo.*", "foo.$foo")                 // invalid $ value
-	shouldErr("foo.*", "foo.{{wildcard(2)}}")      // Mapping function being passed an out of range wildcard index
-	shouldErr("foo.*", "foo.{{unimplemented(1)}}") // Mapping trying to use an unknown mapping function
-	shouldErr("foo.*", "foo.{{partition(10)}}")    // Not enough arguments passed to the mapping function
-	shouldErr("foo.*", "foo.{{wildcard(foo)}}")    // Invalid argument passed to the mapping function
-	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) *subjectTransform {
-		t.Helper()
-		tr, err := newSubjectTransform(src, dest)
-		if err != nil {
-			t.Fatalf("Got an error %v for src=%q and dest=%q", err, src, dest)
-		}
-		return tr
-	}
-
-	shouldBeOK("foo", "bar")
-	shouldBeOK("foo.*.bar.*.baz", "req.$2.$1")
-	shouldBeOK("baz.>", "mybaz.>")
-	shouldBeOK("*", "{{splitfromleft(1,1)}}")
-	shouldBeOK("", "prefix.>")
-
-	shouldMatch := func(src, dest, sample, expected string) {
-		t.Helper()
-		tr := shouldBeOK(src, dest)
-		s, err := tr.Match(sample)
-		if err != nil {
-			t.Fatalf("Got an error %v when expecting a match for %q to %q", err, sample, expected)
-		}
-		if s != expected {
-			t.Fatalf("Dest does not match what was expected. Got %q, expected %q", s, expected)
-		}
-	}
-
-	shouldMatch("", "prefix.>", "foo", "prefix.foo")
-	shouldMatch("foo", "bar", "foo", "bar")
-	shouldMatch("foo.*.bar.*.baz", "req.$2.$1", "foo.A.bar.B.baz", "req.B.A")
-	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(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
-}