Repository: nats-io/nats-server

Issue:
Allow clients to provide a required feature level in API calls
### Proposed change

To support ADR-44 we need to have a way for clients to require a certain feature level when making an API call.



### Use case

Today we can request a Paused consumer be created, but this is only supported on servers with feature level 1 and creating consumers on older servers with this setting set will silently ignore it.

In future, for this scenario, we would like to be able to say only accept this API call on servers with a certain feature level.  

We do not anticipate all clients need to understand the full map of feature vs version due to https://github.com/nats-io/nats-server/issues/6539 this should be detected.  But there would be cases where a feature subtly change behaviour between versions and we might wish to assert we are only running on the new server.

### Contribution

_No response_

Hints from issue comments:
@shashank-jain45: Hi, can i begin working on this?
@ripienaar: @shashank-jain45 contributions are very welcome, I think we need to spend some more time specifying how we'd like this to work, you can maybe propose a behaviour on ADR-44

Options are roughly to add a field to all the API requests or to add a Header to the API requests.  So far we've avoided Headers other than for example enabling compression of responses - in other words things that influence the transport of the data not the shape of the data - and maybe in this case a header make sense but have not given it a lot of thought.

Propose an approach on the ADR and lets discuss there, once the design is locked down code can be written

Solution patch:
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 1c9cc3e5..1b89719e 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -72,14 +72,14 @@ jobs:
       - name: Install Go
         uses: actions/setup-go@v5
         with:
-          go-version: stable
+          go-version-file: "go.mod"
 
       - name: Run golangci-lint
         # Use commit hash here to avoid a re-tagging attack, as this is a third-party action
         # Commit 1481404843c368bc19ca9406f87d6e0fc97bdcfd = tag v7.0.0
         uses: golangci/golangci-lint-action@1481404843c368bc19ca9406f87d6e0fc97bdcfd
         with:
-          version: v2.1.2
+          version: v2.3.1
           skip-cache: true
           skip-save-cache: true
           args: --timeout=5m --config=.golangci.yml
diff --git a/.goreleaser.yml b/.goreleaser.yml
index 53daaddf..954ba3ed 100644
--- a/.goreleaser.yml
+++ b/.goreleaser.yml
@@ -21,7 +21,7 @@ builds:
     env:
       # This is the toolchain version we use for releases. To override, set the env var, e.g.:
       # GORELEASER_TOOLCHAIN="go1.22.8" TARGET='linux_amd64' goreleaser build --snapshot --clean --single-target
-      - GOTOOLCHAIN={{ envOrDefault "GORELEASER_TOOLCHAIN" "go1.24.6" }}
+      - GOTOOLCHAIN={{ envOrDefault "GORELEASER_TOOLCHAIN" "go1.25.0" }}
       - GO111MODULE=on
       - CGO_ENABLED=0
     goos:
diff --git a/go.mod b/go.mod
index 8eb271fe..d49b531c 100644
--- a/go.mod
+++ b/go.mod
@@ -1,20 +1,20 @@
 module github.com/nats-io/nats-server/v2
 
-go 1.23.0
+go 1.24.0
 
-toolchain go1.23.12
+toolchain go1.24.6
 
 require (
 	github.com/antithesishq/antithesis-sdk-go v0.4.3-default-no-op
 	github.com/google/go-tpm v0.9.5
 	github.com/klauspost/compress v1.18.0
 	github.com/minio/highwayhash v1.0.3
-	github.com/nats-io/jwt/v2 v2.7.4
+	github.com/nats-io/jwt/v2 v2.8.0
 	github.com/nats-io/nats.go v1.44.0
 	github.com/nats-io/nkeys v0.4.11
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.6.0
-	golang.org/x/crypto v0.40.0
-	golang.org/x/sys v0.34.0
+	golang.org/x/crypto v0.41.0
+	golang.org/x/sys v0.35.0
 	golang.org/x/time v0.12.0
 )
diff --git a/go.sum b/go.sum
index 44c158ee..417ba1c8 100644
--- a/go.sum
+++ b/go.sum
@@ -8,8 +8,8 @@ github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zt
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
 github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
-github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI=
-github.com/nats-io/jwt/v2 v2.7.4/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
+github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
+github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
 github.com/nats-io/nats.go v1.44.0 h1:ECKVrDLdh/kDPV1g0gAQ+2+m2KprqZK5O/eJAyAnH2M=
 github.com/nats-io/nats.go v1.44.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
 github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
@@ -24,11 +24,11 @@ github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMT
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
 go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
-golang.org/x/crypto v0.40.0 h1:r4x+VvoG5Fm+eJcxMaY8CQM7Lb0l1lsmjGBQ6s8BfKM=
-golang.org/x/crypto v0.40.0/go.mod h1:Qr1vMER5WyS2dfPHAlsOj01wgLbsyWtFn/aY+5+ZdxY=
+golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4=
+golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc=
 golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA=
-golang.org/x/sys v0.34.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
+golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE=
 golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
diff --git a/locksordering.txt b/locksordering.txt
index 42dc30cf..530a8e4a 100644
--- a/locksordering.txt
+++ b/locksordering.txt
@@ -40,3 +40,8 @@ but if holding "clMu" or "ddMu", locking the stream lock afterward would violate
 
     stream -> clMu
     stream -> ddMu
+
+The "mset.batches.mu" lock protects the batching state without needing to hold the stream lock.
+If "clMu" is used to commit a batch, it should only be acquired while already holding the batch lock.
+
+    mset.batches.mu -> clMu
diff --git a/server/auth.go b/server/auth.go
index e97a9cb4..2d735c0e 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -25,6 +25,7 @@ import (
 	"net"
 	"net/url"
 	"regexp"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -65,6 +66,7 @@ type NkeyUser struct {
 	Account                *Account            `json:"account,omitempty"`
 	SigningKey             string              `json:"signing_key,omitempty"`
 	AllowedConnectionTypes map[string]struct{} `json:"connection_types,omitempty"`
+	ProxyRequired          bool                `json:"proxy_required,omitempty"`
 }
 
 // User is for multiple accounts/users.
@@ -75,6 +77,7 @@ type User struct {
 	Account                *Account            `json:"account,omitempty"`
 	ConnectionDeadline     time.Time           `json:"connection_deadline,omitempty"`
 	AllowedConnectionTypes map[string]struct{} `json:"connection_types,omitempty"`
+	ProxyRequired          bool                `json:"proxy_required,omitempty"`
 }
 
 // clone performs a deep copy of the User struct, returning a new clone with
@@ -593,10 +596,30 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 		ao   bool // auth override
 	)
 
+	// Little helper that will log the error as a debug statement, set the auth error in
+	// the connection and return false to indicate authentication failure.
+	setProxyAuthError := func(err error) bool {
+		c.Debugf(err.Error())
+		c.setAuthError(err)
+		return false
+	}
+
+	// Indicate if this connection came from a trusted proxy. Note that if
+	// trustedProxy could be false even if the connection is proxied, but it
+	// means that there was no trusted proxy configured.
+	trustedProxy, ok := s.proxyCheck(c, opts)
+	if trustedProxy && !ok {
+		return setProxyAuthError(ErrAuthProxyNotTrusted)
+	}
+
+	var proxyRequired bool
 	// Check if we have auth callouts enabled at the server level or in the bound account.
 	defer func() {
-		// Default reason
-		reason := AuthenticationViolation.String()
+		authErr := c.getAuthError()
+		if authErr == nil {
+			authErr = ErrAuthentication
+		}
+		reason := getAuthErrClosedState(authErr).String()
 		// No-op
 		if juc == nil && opts.AuthCallout == nil {
 			if !authorized {
@@ -656,7 +679,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 		// If we are here we have an auth callout defined and we have failed auth so far
 		// so we will callout to our auth backend for processing.
 		if !skip {
-			authorized, reason = s.processClientOrLeafCallout(c, opts)
+			authorized, reason = s.processClientOrLeafCallout(c, opts, proxyRequired, trustedProxy)
 		}
 		// Check if we are authorized and in the auth callout account, and if so add in deny publish permissions for the auth subject.
 		if authorized {
@@ -774,6 +797,10 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 			c.Debugf("User JWT not valid: %v", err)
 			return false
 		}
+		if proxyRequired = juc.ProxyRequired; proxyRequired && !trustedProxy {
+			s.mu.Unlock()
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		vr := jwt.CreateValidationResults()
 		juc.Validate(vr)
 		if vr.IsBlocking(true) {
@@ -1050,6 +1077,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	}
 
 	if nkey != nil {
+		if proxyRequired = nkey.ProxyRequired; proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		// If we did not match noAuthUser check signature which is required.
 		if nkey.Nkey != noAuthUser {
 			if c.opts.Sig == _EMPTY_ {
@@ -1081,6 +1111,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 		return true
 	}
 	if user != nil {
+		if proxyRequired = user.ProxyRequired; proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		ok = comparePasswords(user.Password, c.opts.Password)
 		// If we are authorized, register the user which will properly setup any permissions
 		// for pub/sub authorizations.
@@ -1091,6 +1124,9 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	}
 
 	if c.kind == CLIENT {
+		if proxyRequired = opts.ProxyRequired; proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		if token != _EMPTY_ {
 			return comparePasswords(token, c.opts.Token)
 		} else if username != _EMPTY_ {
@@ -1103,6 +1139,61 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	return false
 }
 
+// If there are configured trusted proxies and this connection comes
+// from a proxy whose signature can be verified by one of the known
+// trusted key, this function will return `true, true`. If the signature
+// cannot be verified by any, it will return `true, false`.
+// If the connectio is not proxied, or there are no configured trusted
+// proxies, then this function returns `false, false`.
+//
+// Server lock MUST NOT be held on entry since this function will grab
+// the read lock to extract the list of proxy trusted keys. The signature
+// verification process will be done outside of the lock.
+func (s *Server) proxyCheck(c *client, opts *Options) (bool, bool) {
+	// If there is no signature or no configured trusted proxy, return false.
+	psig := c.opts.ProxySig
+	if psig == _EMPTY_ || opts.Proxies == nil || len(opts.Proxies.Trusted) == 0 {
+		return false, false
+	}
+	// Decode the signature.
+	sig, err := base64.RawURLEncoding.DecodeString(psig)
+	if err != nil {
+		c.Debugf("Proxy signature not valid base64")
+		return true, false
+	}
+	// Go through the trusted keys and verify the signature.
+	s.mu.RLock()
+	keys := slices.Clone(s.proxiesKeyPairs)
+	s.mu.RUnlock()
+	for _, kp := range keys {
+		// We stop at the first that is valid.
+		if err := kp.Verify(c.nonce, sig); err == nil {
+			pub, _ := kp.PublicKey()
+			// Track which proxy public key is used by this connection.
+			c.mu.Lock()
+			c.proxyKey = pub
+			cid := c.cid
+			c.mu.Unlock()
+			// Track this proxied connection so that it can be closed
+			// if the trusted key is removed on configuration reload.
+			s.mu.Lock()
+			if s.proxiedConns == nil {
+				s.proxiedConns = make(map[string]map[uint64]*client)
+			}
+			clients := s.proxiedConns[pub]
+			if clients == nil {
+				clients = make(map[uint64]*client)
+			}
+			clients[cid] = c
+			s.proxiedConns[pub] = clients
+			s.mu.Unlock()
+			return true, true
+		}
+	}
+	// We could not verify the signature, so indicate failure.
+	return true, false
+}
+
 func getTLSAuthDCs(rdns *pkix.RDNSequence) string {
 	dcOID := asn1.ObjectIdentifier{0, 9, 2342, 19200300, 100, 1, 25}
 	dcs := []string{}
@@ -1351,7 +1442,25 @@ func (s *Server) registerLeafWithAccount(c *client, account string) bool {
 func (s *Server) isLeafNodeAuthorized(c *client) bool {
 	opts := s.getOpts()
 
-	isAuthorized := func(username, password, account string) bool {
+	setProxyAuthError := func(err error) bool {
+		c.Debugf(err.Error())
+		c.setAuthError(err)
+		return false
+	}
+
+	isAuthorized := func(username, password, account string, proxyRequired bool) bool {
+		trustedProxy, ok := s.proxyCheck(c, opts)
+		if trustedProxy && !ok {
+			return setProxyAuthError(ErrAuthProxyNotTrusted)
+		}
+		// A given user may not be required, but if the boolean is set at the
+		// authorization top-level, then override.
+		if !proxyRequired && opts.LeafNode.ProxyRequired {
+			proxyRequired = true
+		}
+		if proxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		if username != c.opts.Username {
 			return false
 		}
@@ -1365,8 +1474,16 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 	// The user in CONNECT must match. We will bind to the account associated
 	// with that user (from the leafnode's authorization{} config).
 	if opts.LeafNode.Username != _EMPTY_ {
-		return isAuthorized(opts.LeafNode.Username, opts.LeafNode.Password, opts.LeafNode.Account)
+		return isAuthorized(opts.LeafNode.Username, opts.LeafNode.Password, opts.LeafNode.Account,
+			opts.LeafNode.ProxyRequired)
 	} else if opts.LeafNode.Nkey != _EMPTY_ {
+		trustedProxy, ok := s.proxyCheck(c, opts)
+		if trustedProxy && !ok {
+			return false
+		}
+		if opts.LeafNode.ProxyRequired && !trustedProxy {
+			return setProxyAuthError(ErrAuthProxyRequired)
+		}
 		if c.opts.Nkey != opts.LeafNode.Nkey {
 			return false
 		}
@@ -1420,7 +1537,7 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 			}
 			// This will authorize since are using an existing user,
 			// but it will also register with proper account.
-			return isAuthorized(user.Username, user.Password, accName)
+			return isAuthorized(user.Username, user.Password, accName, user.ProxyRequired)
 		}
 
 		// This is expected to be a very small array.
@@ -1430,7 +1547,7 @@ func (s *Server) isLeafNodeAuthorized(c *client) bool {
 				if u.Account != nil {
 					accName = u.Account.Name
 				}
-				return isAuthorized(u.Username, u.Password, accName)
+				return isAuthorized(u.Username, u.Password, accName, u.ProxyRequired)
 			}
 		}
 		return false
@@ -1535,3 +1652,45 @@ func validateNoAuthUser(o *Options, noAuthUser string) error {
 		`no_auth_user: "%s" not present as user or nkey in authorization block or account configuration`,
 		noAuthUser)
 }
+
+func validateProxies(o *Options) error {
+	if o.Proxies == nil {
+		return nil
+	}
+	for _, p := range o.Proxies.Trusted {
+		if !nkeys.IsValidPublicKey(p.Key) {
+			return fmt.Errorf("proxy trusted key %q is invalid", p.Key)
+		}
+	}
+	return nil
+}
+
+// Create a list of nkeys.KeyPair corresponding to the public keys
+// of the Proxies.TrustedKeys list.
+// Server lock must be held on entry.
+func (s *Server) processProxiesTrustedKeys() {
+	// We could be here on reload.
+	if s.proxiesKeyPairs != nil {
+		s.proxiesKeyPairs = s.proxiesKeyPairs[:0]
+	}
+	if opts := s.getOpts(); opts.Proxies == nil {
+		return
+	}
+	for _, p := range s.getOpts().Proxies.Trusted {
+		// Can't fail since we have already checked that it was a valid key.
+		kp, _ := nkeys.FromPublicKey(p.Key)
+		s.proxiesKeyPairs = append(s.proxiesKeyPairs, kp)
+	}
+}
+
+// Returns the connection's `ClosedState` for the given authenication error.
+func getAuthErrClosedState(authErr error) ClosedState {
+	switch authErr {
+	case ErrAuthProxyNotTrusted:
+		return ProxyNotTrusted
+	case ErrAuthProxyRequired:
+		return ProxyRequired
+	default:
+		return AuthenticationViolation
+	}
+}
diff --git a/server/auth_callout.go b/server/auth_callout.go
index dded0626..a7ed073e 100644
--- a/server/auth_callout.go
+++ b/server/auth_callout.go
@@ -33,7 +33,7 @@ const (
 )
 
 // Process a callout on this client's behalf.
-func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorized bool, errStr string) {
+func (s *Server) processClientOrLeafCallout(c *client, opts *Options, proxyRequired, trustedProxy bool) (authorized bool, errStr string) {
 	isOperatorMode := len(opts.TrustedKeys) > 0
 
 	// this is the account the user connected in, or the one running the callout
@@ -245,6 +245,16 @@ func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorize
 			respCh <- titleCase(err.Error())
 			return
 		}
+		// If the caller had established that the user should go through a proxy,
+		// or if the `arc` JWT requires it, and we don't have a trusted proxy,
+		// reject the connection.
+		if (proxyRequired || arc.ProxyRequired) && !trustedProxy {
+			err = ErrAuthProxyRequired
+			c.setAuthError(err)
+			c.authViolation()
+			respCh <- titleCase(err.Error())
+			return
+		}
 		vr := jwt.CreateValidationResults()
 		arc.Validate(vr)
 		if len(vr.Issues) > 0 {
diff --git a/server/client.go b/server/client.go
index 14f324bf..c6e8e5d4 100644
--- a/server/client.go
+++ b/server/client.go
@@ -223,6 +223,8 @@ const (
 	MinimumVersionRequired
 	ClusterNamesIdentical
 	Kicked
+	ProxyNotTrusted
+	ProxyRequired
 )
 
 // Some flags passed to processMsgResults
@@ -271,6 +273,7 @@ type client struct {
 	msgb       [msgScratchSize]byte
 	last       time.Time
 	lastIn     time.Time
+	proxyKey   string
 
 	repliesSincePrune uint16
 	lastReplyPrune    time.Time
@@ -299,6 +302,13 @@ type client struct {
 	nameTag string
 
 	tlsTo *time.Timer
+
+	// Authentication error override. This is used because the authentication
+	// stack is simply returning a boolean, and the only authentication error
+	// reported is the generic `ErrAuthentication`. In the authentication code,
+	// if we want to report a different error, we can now set this field
+	// and `authViolation()` will use that one.
+	authErr error
 }
 
 type rrTracking struct {
@@ -653,6 +663,9 @@ type ClientOpts struct {
 
 	// Leafnodes
 	RemoteAccount string `json:"remote_account,omitempty"`
+
+	// Proxy would include its own nonce signature.
+	ProxySig string `json:"proxy_sig,omitempty"`
 }
 
 var defaultOpts = ClientOpts{Verbose: true, Pedantic: true, Echo: true}
@@ -2277,6 +2290,12 @@ func (c *client) accountAuthExpired() {
 }
 
 func (c *client) authViolation() {
+	authErr := c.getAuthError()
+	if authErr == nil {
+		authErr = ErrAuthentication
+	}
+	reason := getAuthErrClosedState(authErr)
+
 	var s *Server
 	var hasTrustedNkeys, hasNkeys, hasUsers bool
 	if s = c.srv; s != nil {
@@ -2285,30 +2304,31 @@ func (c *client) authViolation() {
 		hasNkeys = s.nkeys != nil
 		hasUsers = s.users != nil
 		s.mu.RUnlock()
-		defer s.sendAuthErrorEvent(c)
+		defer s.sendAuthErrorEvent(c, reason.String())
 	}
 
 	if hasTrustedNkeys {
-		c.Errorf("%v", ErrAuthentication)
+		c.Errorf("%v", authErr)
 	} else if hasNkeys {
 		c.Errorf("%s - Nkey %q",
-			ErrAuthentication.Error(),
+			authErr.Error(),
 			c.opts.Nkey)
 	} else if hasUsers {
 		c.Errorf("%s - User %q",
-			ErrAuthentication.Error(),
+			authErr.Error(),
 			c.opts.Username)
 	} else {
 		if c.srv != nil {
-			c.Errorf(ErrAuthentication.Error())
+			c.Errorf(authErr.Error())
 		}
 	}
 	if c.isMqtt() {
 		c.mqttEnqueueConnAck(mqttConnAckRCNotAuthorized, false)
 	} else {
+		// Send this to client, regardless of the authErr override.
 		c.sendErr("Authorization Violation")
 	}
-	c.closeConnection(AuthenticationViolation)
+	c.closeConnection(reason)
 }
 
 func (c *client) maxAccountConnExceeded() {
@@ -6398,3 +6418,18 @@ func (c *client) setFirstPingTimer() {
 	}
 	c.ping.tmr = time.AfterFunc(d, c.processPingTimer)
 }
+
+// Sets this error as the authentication error. To be used in authViolation()
+// to report an error different of `ErrAuthentication`.
+func (c *client) setAuthError(err error) {
+	c.mu.Lock()
+	c.authErr = err
+	c.mu.Unlock()
+}
+
+// Returns the authentication error set in the connection, possibly nil.
+func (c *client) getAuthError() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.authErr
+}
diff --git a/server/const.go b/server/const.go
index 9f51591d..8185ba35 100644
--- a/server/const.go
+++ b/server/const.go
@@ -58,7 +58,7 @@ func init() {
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.12.0-dev"
+	VERSION = "2.12.0-preview.1"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 661235ce..dceb7df4 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math"
 	"math/rand"
 	"reflect"
 	"regexp"
@@ -115,7 +116,7 @@ type ConsumerConfig struct {
 	// Force memory storage.
 	MemoryStorage bool `json:"mem_storage,omitempty"`
 	// Force the consumer to only deliver messages if the stream has at minimum this specified last sequence.
-	MinLastSeq uint64 `json:"min_last_seq,omitempty"`
+	MinLastSeq uint64 `json:"-"`
 
 	// Don't add to general clients.
 	Direct bool `json:"direct,omitempty"`
@@ -220,18 +221,22 @@ const (
 	PriorityOverflow
 	// Single client takes over handling of the messages, while others are on standby.
 	PriorityPinnedClient
+	// Clients with lowest priority will be selected first.
+	PriorityPrioritized
 )
 
 const (
 	PriorityNoneJSONString         = `"none"`
 	PriorityOverflowJSONString     = `"overflow"`
 	PriorityPinnedClientJSONString = `"pinned_client"`
+	PriorityPrioritizedJSONString  = `"prioritized"`
 )
 
 var (
 	PriorityNoneJSONBytes         = []byte(PriorityNoneJSONString)
 	PriorityOverflowJSONBytes     = []byte(PriorityOverflowJSONString)
 	PriorityPinnedClientJSONBytes = []byte(PriorityPinnedClientJSONString)
+	PriorityPrioritizedJSONBytes  = []byte(PriorityPrioritizedJSONString)
 )
 
 func (pp PriorityPolicy) String() string {
@@ -240,6 +245,8 @@ func (pp PriorityPolicy) String() string {
 		return PriorityOverflowJSONString
 	case PriorityPinnedClient:
 		return PriorityPinnedClientJSONString
+	case PriorityPrioritized:
+		return PriorityPrioritizedJSONString
 	default:
 		return PriorityNoneJSONString
 	}
@@ -251,6 +258,8 @@ func (pp PriorityPolicy) MarshalJSON() ([]byte, error) {
 		return PriorityOverflowJSONBytes, nil
 	case PriorityPinnedClient:
 		return PriorityPinnedClientJSONBytes, nil
+	case PriorityPrioritized:
+		return PriorityPrioritizedJSONBytes, nil
 	case PriorityNone:
 		return PriorityNoneJSONBytes, nil
 	default:
@@ -264,6 +273,8 @@ func (pp *PriorityPolicy) UnmarshalJSON(data []byte) error {
 		*pp = PriorityOverflow
 	case PriorityPinnedClientJSONString:
 		*pp = PriorityPinnedClient
+	case PriorityPrioritizedJSONString:
+		*pp = PriorityPrioritized
 	case PriorityNoneJSONString:
 		*pp = PriorityNone
 	default:
@@ -3454,6 +3465,7 @@ type PriorityGroup struct {
 	MinPending    int64  `json:"min_pending,omitempty"`
 	MinAckPending int64  `json:"min_ack_pending,omitempty"`
 	Id            string `json:"id,omitempty"`
+	Priority      int    `json:"priority,omitempty"`
 }
 
 // Used in nextReqFromMsg, since the json.Unmarshal causes the request
@@ -3583,6 +3595,33 @@ var (
 	errWaitQueueNil  = errors.New("wait queue is nil")
 )
 
+// insertSorted inserts wr at the correct position based on priority
+func (wq *waitQueue) insertSorted(wr *waitingRequest) {
+
+	// Handle empty queue
+	if wq.head == nil {
+		wq.head = wr
+		wq.tail = wr
+		wr.next = nil
+		return
+	}
+	insertAtPosition(wr, wq)
+}
+
+func (wq *waitQueue) addPrioritized(wr *waitingRequest) error {
+	if wq == nil {
+		return errWaitQueueNil
+	}
+	if wq.isFull() {
+		return errWaitQueueFull
+	}
+
+	wq.insertSorted(wr)
+	wq.n++
+	wq.last = wr.received
+	return nil
+}
+
 // Adds in a new request.
 func (wq *waitQueue) add(wr *waitingRequest) error {
 	if wq == nil {
@@ -3646,6 +3685,17 @@ func (wq *waitQueue) cycle() {
 	}
 }
 
+func (wq *waitQueue) popOrPopAndRequeue(priority PriorityPolicy) *waitingRequest {
+	if wq == nil || wq.head == nil {
+		return nil
+	}
+
+	if priority == PriorityPrioritized {
+		return wq.popAndRequeue()
+	}
+	return wq.pop()
+}
+
 // pop will return the next request and move the read cursor.
 // This will now place a request that still has pending items at the ends of the list.
 func (wq *waitQueue) pop() *waitingRequest {
@@ -3665,6 +3715,74 @@ func (wq *waitQueue) pop() *waitingRequest {
 	return wr
 }
 
+// popAndRequeue pops the head element and requeues it at the end of its priority group.
+// This maintains FIFO order within the same priority level.
+func (wq *waitQueue) popAndRequeue() *waitingRequest {
+	if wq == nil || wq.head == nil {
+		return nil
+	}
+
+	// Save the head
+	wr := wq.head
+
+	if wr == nil {
+		return wr
+	}
+
+	wr.d++
+	wr.n--
+
+	if wr.n > 0 && wq.n > 1 {
+		if wr.next == nil {
+			return wr
+		}
+		wq.head = wq.head.next
+		wr.next = nil
+
+		insertAtPosition(wr, wq)
+
+	} else if wr.n <= 0 {
+		wq.removeCurrent()
+		return wr
+
+	}
+
+	return wr
+}
+
+func insertAtPosition(wr *waitingRequest, wq *waitQueue) {
+	priority := math.MaxInt32
+	if wr.priorityGroup != nil {
+		priority = wr.priorityGroup.Priority
+	}
+
+	var prev *waitingRequest
+	current := wq.head
+	for current != nil {
+		currentPriority := math.MaxInt32
+		if current.priorityGroup != nil {
+			currentPriority = current.priorityGroup.Priority
+		}
+		if currentPriority > priority {
+			break
+		}
+		prev = current
+		current = current.next
+	}
+
+	if prev == nil {
+		// All remaining elements have higher priority
+		wr.next = wq.head
+		wq.head = wr
+	} else {
+		wr.next = prev.next
+		prev.next = wr
+		if wr.next == nil {
+			wq.tail = wr
+		}
+	}
+}
+
 // Removes the current read pointer (head FIFO) entry.
 func (wq *waitQueue) removeCurrent() {
 	wq.remove(nil, wq.head)
@@ -3826,17 +3944,17 @@ func (o *consumer) nextWaiting(sz int) *waitingRequest {
 				if needNewPin {
 					o.sendPinnedAdvisoryLocked(priorityGroup)
 				}
-				return o.waiting.pop()
+				return o.waiting.popOrPopAndRequeue(o.cfg.PriorityPolicy)
 			} else if time.Since(wr.received) < defaultGatewayRecentSubExpiration && (o.srv.leafNodeEnabled || o.srv.gateway.enabled) {
 				if needNewPin {
 					o.sendPinnedAdvisoryLocked(priorityGroup)
 				}
-				return o.waiting.pop()
+				return o.waiting.popOrPopAndRequeue(o.cfg.PriorityPolicy)
 			} else if o.srv.gateway.enabled && o.srv.hasGatewayInterest(wr.acc.Name, wr.interest) {
 				if needNewPin {
 					o.sendPinnedAdvisoryLocked(priorityGroup)
 				}
-				return o.waiting.pop()
+				return o.waiting.popOrPopAndRequeue(o.cfg.PriorityPolicy)
 			}
 		} else {
 			// We do check for expiration in `processWaiting`, but it is possible to hit the expiry here, and not there.
@@ -3911,7 +4029,7 @@ func (nmr *nextMsgReq) returnToPool() {
 // processNextMsgReq will process a request for the next message available. A nil message payload means deliver
 // a single message. If the payload is a formal request or a number parseable with Atoi(), then we will send a
 // batch of messages without requiring another request to this endpoint, or an ACK.
-func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _, reply string, msg []byte) {
+func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _, reply string, rmsg []byte) {
 	if reply == _EMPTY_ {
 		return
 	}
@@ -3923,7 +4041,12 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 		return
 	}
 
-	_, msg = c.msgParts(msg)
+	hdr, msg := c.msgParts(rmsg)
+	if errorOnRequiredApiLevel(hdr) {
+		hdr = []byte("NATS/1.0 412 Required Api Level\r\n\r\n")
+		o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+		return
+	}
 	o.nextMsgReqs.push(newNextMsgReq(reply, copyBytes(msg)))
 }
 
@@ -3977,6 +4100,10 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 		if priorityGroup.Id != _EMPTY_ && o.cfg.PriorityPolicy != PriorityPinnedClient {
 			sendErr(400, "Bad Request - Not a Pinned Client Priority consumer")
 		}
+		if priorityGroup.Priority < 0 || priorityGroup.Priority > 9 {
+			sendErr(400, "Bad Request - Priority must be between 0 and 9")
+			return
+		}
 	}
 
 	if priorityGroup != nil && o.cfg.PriorityPolicy != PriorityNone {
@@ -4047,15 +4174,25 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 	wr.b = maxBytes
 	wr.received = time.Now()
 
-	if err := o.waiting.add(wr); err != nil {
-		// If the client has a heartbeat interval set, don't bother responding with a 409,
-		// otherwise we can end up in a hot loop with the client re-requesting instead of
-		// waiting for the missing heartbeats instead and retrying.
-		if hb == 0 {
-			sendErr(409, "Exceeded MaxWaiting")
+	if o.cfg.PriorityPolicy == PriorityPrioritized {
+		if err := o.waiting.addPrioritized(wr); err != nil {
+			if hb == 0 {
+				sendErr(409, "Exceeded MaxWaiting")
+			}
+			wr.recycle()
+			return
+		}
+	} else {
+		if err := o.waiting.add(wr); err != nil {
+			// If the client has a heartbeat interval set, don't bother responding with a 409,
+			// otherwise we can end up in a hot loop with the client re-requesting instead of
+			// waiting for the missing heartbeats instead and retrying.
+			if hb == 0 {
+				sendErr(409, "Exceeded MaxWaiting")
+			}
+			wr.recycle()
+			return
 		}
-		wr.recycle()
-		return
 	}
 	o.signalNewMessages()
 	// If we are clustered update our followers about this request.
diff --git a/server/disk_avail.go b/server/disk_avail.go
index b879330e..e27c1a71 100644
--- a/server/disk_avail.go
+++ b/server/disk_avail.go
@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build !windows && !openbsd && !netbsd && !wasm
+//go:build !windows && !openbsd && !netbsd && !wasm && !illumos && !solaris
 
 package server
 
diff --git a/server/disk_avail_solaris.go b/server/disk_avail_solaris.go
new file mode 100644
index 00000000..e0febd1f
--- /dev/null
+++ b/server/disk_avail_solaris.go
@@ -0,0 +1,38 @@
+// Copyright 2025 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build illumos || solaris
+
+package server
+
+import (
+	"os"
+	"golang.org/x/sys/unix"
+)
+
+func diskAvailable(storeDir string) int64 {
+	var ba int64
+	if _, err := os.Stat(storeDir); os.IsNotExist(err) {
+		os.MkdirAll(storeDir, defaultDirPerms)
+	}
+	var fs unix.Statvfs_t
+	if err := unix.Statvfs(storeDir, &fs); err == nil {
+		// Estimate 75% of available storage.
+		ba = int64(uint64(fs.Frsize) * uint64(fs.Bavail) / 4 * 3)
+	} else {
+		// Used 1TB default as a guess if all else fails.
+		ba = JetStreamMaxStoreDefault
+	}
+	return ba
+}
+
diff --git a/server/errors.go b/server/errors.go
index 1a83abc4..07a3ca69 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -31,6 +31,14 @@ var (
 	// ErrAuthExpired represents an expired authorization due to timeout.
 	ErrAuthExpired = errors.New("authentication expired")
 
+	// ErrAuthProxyNotTrusted represents an error condition on failed authentication
+	// due to a connection from a proxy not in the list of trusted proxies.
+	ErrAuthProxyNotTrusted = errors.New("proxy is not trusted")
+
+	// ErrAuthProxyRequired represents an error condition on failed authentication
+	// due to a connection not coming from a proxy.
+	ErrAuthProxyRequired = errors.New("proxy connection required")
+
 	// ErrMaxPayload represents an error condition when the payload is too big.
 	ErrMaxPayload = errors.New("maximum payload exceeded")
 
diff --git a/server/errors.json b/server/errors.json
index 94c3b863..e5ff2fa1 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1828,5 +1828,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSRequiredApiLevelErr",
+    "code": 412,
+    "error_code": 10185,
+    "description": "JetStream minimum api level required",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/events.go b/server/events.go
index f78303bf..de588047 100644
--- a/server/events.go
+++ b/server/events.go
@@ -2610,7 +2610,7 @@ func (s *Server) accountDisconnectEvent(c *client, now time.Time, reason string)
 }
 
 // This is the system level event sent to the system account for operators.
-func (s *Server) sendAuthErrorEvent(c *client) {
+func (s *Server) sendAuthErrorEvent(c *client, reason string) {
 	s.mu.Lock()
 	if !s.eventsEnabled() {
 		s.mu.Unlock()
@@ -2658,7 +2658,7 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 				Bytes: c.outBytes,
 			},
 		},
-		Reason: AuthenticationViolation.String(),
+		Reason: reason,
 	}
 	c.mu.Unlock()
 
diff --git a/server/filestore.go b/server/filestore.go
index f98eb641..06904fa4 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -286,6 +286,8 @@ const (
 	hdrLen = 2
 	// This is where we keep the streams.
 	streamsDir = "streams"
+	// This is where we keep inflight batches for streams.
+	batchesDir = "batches"
 	// This is where we keep the message store blocks.
 	msgDir = "msgs"
 	// This is where we temporarily move the messages dir.
@@ -677,7 +679,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 
 	if lmb := fs.lmb; lmb != nil {
 		// Enable/disable async flush depending on if it's supported and already initialized.
-		supportsAsyncFlush := cfg.AllowAsyncFlush && cfg.Replicas > 1
+		supportsAsyncFlush := !fs.fcfg.SyncAlways && cfg.Replicas > 1
 		if supportsAsyncFlush && !fs.fcfg.AsyncFlush {
 			fs.fcfg.AsyncFlush = true
 			lmb.spinUpFlushLoop()
@@ -5078,7 +5080,7 @@ func (mb *msgBlock) compactWithFloor(floor uint64) {
 // Grab info from a slot.
 // Lock should be held.
 func (mb *msgBlock) slotInfo(slot int) (uint32, uint32, bool, error) {
-	if mb.cache == nil || slot >= len(mb.cache.idx) {
+	if slot < 0 || mb.cache == nil || slot >= len(mb.cache.idx) {
 		return 0, 0, false, errPartialCache
 	}
 
@@ -8680,6 +8682,10 @@ func (fs *fileStore) Truncate(seq uint64) error {
 		return ErrStoreSnapshotInProgress
 	}
 
+	// Any existing state file will no longer be applicable. We will force write a new one
+	// at the end, after we release the lock.
+	os.Remove(filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile))
+
 	var lsm *StoreMsg
 	smb := fs.selectMsgBlock(seq)
 	if smb != nil {
@@ -8706,9 +8712,15 @@ func (fs *fileStore) Truncate(seq uint64) error {
 		return err
 	}
 
+	// The selected message block needs to be removed if it needs to be fully truncated.
+	var removeSmb bool
+	if smb != nil {
+		removeSmb = atomic.LoadUint64(&smb.first.seq) > seq
+	}
+
 	// If the selected block is not found or the message was deleted, we'll need to write a tombstone
 	// at the truncated sequence so we don't roll backward on our last sequence and timestamp.
-	if lsm == nil {
+	if lsm == nil || removeSmb {
 		fs.writeTombstone(seq, lastTime)
 	}
 
@@ -8748,8 +8760,28 @@ func (fs *fileStore) Truncate(seq uint64) error {
 
 	hasWrittenTombstones := len(tmb.tombs()) > 0
 	if smb != nil {
-		// Make sure writeable.
 		smb.mu.Lock()
+		if removeSmb {
+			purged += smb.msgs
+			bytes += smb.bytes
+
+			// We could have tombstones for messages before the truncated sequence.
+			if tombs := smb.tombsLocked(); len(tombs) > 0 {
+				// Temporarily unlock while we write tombstones.
+				smb.mu.Unlock()
+				for _, tomb := range tombs {
+					if tomb.seq < seq {
+						fs.writeTombstone(tomb.seq, tomb.ts)
+					}
+				}
+				smb.mu.Lock()
+			}
+			fs.removeMsgBlock(smb)
+			smb.mu.Unlock()
+			goto SKIP
+		}
+
+		// Make sure writeable.
 		if err := smb.enableForWriting(fs.fip); err != nil {
 			smb.mu.Unlock()
 			fs.mu.Unlock()
@@ -8780,6 +8812,7 @@ func (fs *fileStore) Truncate(seq uint64) error {
 		smb.mu.Unlock()
 	}
 
+SKIP:
 	// If no tombstones were written, we can remove the block and
 	// purely rely on the selected block as the last block.
 	if !hasWrittenTombstones {
@@ -8805,9 +8838,6 @@ func (fs *fileStore) Truncate(seq uint64) error {
 	// Reset our subject lookup info.
 	fs.resetGlobalPerSubjectInfo()
 
-	// Any existing state file no longer applicable. We will force write a new one
-	// after we release the lock.
-	os.Remove(filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile))
 	fs.dirty++
 
 	cb := fs.scb
@@ -9252,7 +9282,7 @@ func (fs *fileStore) closeAllMsgBlocks(sync bool) {
 	}
 }
 
-func (fs *fileStore) Delete() error {
+func (fs *fileStore) Delete(inline bool) error {
 	if fs.isClosed() {
 		// Always attempt to remove since we could have been closed beforehand.
 		os.RemoveAll(fs.fcfg.StoreDir)
@@ -9307,7 +9337,7 @@ func (fs *fileStore) Delete() error {
 	}
 	// Do this in separate Go routine in case lots of blocks.
 	// Purge above protects us as does the removal of meta artifacts above.
-	go func() {
+	removeDir := func() {
 		<-dios
 		err := os.RemoveAll(ndir)
 		dios <- struct{}{}
@@ -9324,8 +9354,12 @@ func (fs *fileStore) Delete() error {
 				return
 			}
 		}
-	}()
-
+	}
+	if inline {
+		removeDir()
+	} else {
+		go removeDir()
+	}
 	return nil
 }
 
diff --git a/server/jetstream.go b/server/jetstream.go
index 46993b2b..76e5c308 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1404,6 +1404,61 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			mset.setCreatedTime(cfg.Created)
 		}
 
+		// Might need to recover from a partial batch write.
+		if cfg.AllowAtomicPublish {
+			var (
+				ok            bool
+				smv           StoreMsg
+				batchId       string
+				batchSeq      uint64
+				commit        bool
+				batchStoreDir string
+				store         StreamStore
+				state         StreamState
+			)
+			// Check if the last message was part of a batch.
+			sm, err := mset.store.LoadLastMsg(fwcs, &smv)
+			if err != nil || sm == nil {
+				goto SKIP
+			}
+			batchId = getBatchId(sm.hdr)
+			batchSeq, ok = getBatchSequence(sm.hdr)
+			commit = len(sliceHeader(JSBatchCommit, sm.hdr)) != 0
+			if batchId == _EMPTY_ || !ok || commit {
+				goto SKIP
+			}
+			// We've observed a partial batch write. Write the remainder of the batch.
+			batchSeq++
+			_, batchStoreDir = getBatchStoreDir(mset, batchId)
+			if _, err = os.Stat(batchStoreDir); err != nil {
+				s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: %v",
+					mset.accName(), mset.name(), batchSeq, err)
+				goto SKIP
+			}
+			store, err = newBatchStore(mset, batchId)
+			if err != nil {
+				s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: %v",
+					mset.accName(), mset.name(), batchSeq, err)
+				goto SKIP
+			}
+			store.FastState(&state)
+			s.Noticef("  Restoring partial batch write for stream '%s > %s' (seq %d to %d)",
+				mset.accName(), mset.name(), batchSeq, state.LastSeq)
+			// Loop through items that weren't persisted yet.
+			for seq := batchSeq; seq <= state.LastSeq; seq++ {
+				sm, err = store.LoadMsg(seq, &smv)
+				if err != nil || sm == nil {
+					s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: %v",
+						mset.accName(), mset.name(), seq, err)
+					break
+				}
+				mset.processJetStreamMsg(sm.subj, _EMPTY_, sm.hdr, sm.msg, 0, 0, nil, false)
+			}
+			store.Delete(true)
+		SKIP:
+			os.RemoveAll(filepath.Join(sdir, fi.Name(), batchesDir))
+		}
+
 		state := mset.state()
 		s.Noticef("  Restored %s messages for stream '%s > %s' in %v",
 			comma(int64(state.Msgs)), mset.accName(), mset.name(), time.Since(rt).Round(time.Millisecond))
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 69da6a49..42c2b1b2 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -333,6 +333,12 @@ const (
 	JSAuditAdvisory = "$JS.EVENT.ADVISORY.API"
 )
 
+// Headers used in $JS.API.> requests.
+const (
+	// JSRequiredApiLevel requires the API level of the responding server to have the specified minimum value.
+	JSRequiredApiLevel = "Nats-Required-Api-Level"
+)
+
 var denyAllClientJs = []string{jsAllAPI, "$KV.>", "$OBJ.>"}
 var denyAllJs = []string{jscAllSubj, raftAllSubj, jsAllAPI, "$KV.>", "$OBJ.>"}
 
@@ -680,7 +686,7 @@ type JSApiMsgGetRequest struct {
 	NextFor string `json:"next_by_subj,omitempty"`
 
 	// Force the server to only deliver messages if the stream has at minimum this specified last sequence.
-	MinLastSeq uint64 `json:"min_last_seq,omitempty"`
+	MinLastSeq uint64 `json:"-"`
 
 	// Batch support. Used to request more than one msg at a time.
 	// Can be used with simple starting seq, but also NextFor with wildcards.
@@ -1285,13 +1291,18 @@ func (s *Server) jsAccountInfoRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiAccountInfoResponse{ApiResponse: ApiResponse{Type: JSApiAccountInfoResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -1346,13 +1357,18 @@ func (s *Server) jsTemplateCreateRequest(sub *subscription, c *client, _ *Accoun
 	if c == nil {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamTemplateCreateResponse{ApiResponse: ApiResponse{Type: JSApiStreamTemplateCreateResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if !acc.JetStreamEnabled() {
 		resp.Error = NewJSNotEnabledForAccountError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1402,13 +1418,18 @@ func (s *Server) jsTemplateNamesRequest(sub *subscription, c *client, _ *Account
 	if c == nil {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamTemplateNamesResponse{ApiResponse: ApiResponse{Type: JSApiStreamTemplateNamesResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if !acc.JetStreamEnabled() {
 		resp.Error = NewJSNotEnabledForAccountError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1467,13 +1488,18 @@ func (s *Server) jsTemplateInfoRequest(sub *subscription, c *client, _ *Account,
 	if c == nil {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamTemplateInfoResponse{ApiResponse: ApiResponse{Type: JSApiStreamTemplateInfoResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if !acc.JetStreamEnabled() {
 		resp.Error = NewJSNotEnabledForAccountError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1509,13 +1535,18 @@ func (s *Server) jsTemplateDeleteRequest(sub *subscription, c *client, _ *Accoun
 	if c == nil {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamTemplateDeleteResponse{ApiResponse: ApiResponse{Type: JSApiStreamTemplateDeleteResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if !acc.JetStreamEnabled() {
 		resp.Error = NewJSNotEnabledForAccountError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1576,13 +1607,18 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamCreateResponse{ApiResponse: ApiResponse{Type: JSApiStreamCreateResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -1688,13 +1724,18 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -1781,13 +1822,18 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, _ *Account,
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamNamesResponse{ApiResponse: ApiResponse{Type: JSApiStreamNamesResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -1908,7 +1954,7 @@ func (s *Server) jsStreamListRequest(sub *subscription, c *client, _ *Account, s
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -1918,6 +1964,11 @@ func (s *Server) jsStreamListRequest(sub *subscription, c *client, _ *Account, s
 		ApiResponse: ApiResponse{Type: JSApiStreamListResponseType},
 		Streams:     []*StreamInfo{},
 	}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -2025,6 +2076,11 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 	if rt := getHeader(JSResponseType, hdr); len(rt) > 0 && string(rt) == jsCreateResponse {
 		resp.ApiResponse.Type = JSApiStreamCreateResponseType
 	}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	var clusterWideConsCount int
 
@@ -2216,7 +2272,7 @@ func (s *Server) jsStreamLeaderStepDownRequest(sub *subscription, c *client, _ *
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -2226,6 +2282,11 @@ func (s *Server) jsStreamLeaderStepDownRequest(sub *subscription, c *client, _ *
 	name := tokenAt(subject, 6)
 
 	var resp = JSApiStreamLeaderStepDownResponse{ApiResponse: ApiResponse{Type: JSApiStreamLeaderStepDownResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// If we are not in clustered mode this is a failed request.
 	if !s.JetStreamIsClustered() {
@@ -2326,13 +2387,18 @@ func (s *Server) jsConsumerLeaderStepDownRequest(sub *subscription, c *client, _
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiConsumerLeaderStepDownResponse{ApiResponse: ApiResponse{Type: JSApiConsumerLeaderStepDownResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// If we are not in clustered mode this is a failed request.
 	if !s.JetStreamIsClustered() {
@@ -2444,7 +2510,7 @@ func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, _ *Acco
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -2454,6 +2520,11 @@ func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, _ *Acco
 	name := tokenAt(subject, 6)
 
 	var resp = JSApiStreamRemovePeerResponse{ApiResponse: ApiResponse{Type: JSApiStreamRemovePeerResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// If we are not in clustered mode this is a failed request.
 	if !s.JetStreamIsClustered() {
@@ -2547,7 +2618,7 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -2571,6 +2642,11 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 	}
 
 	var resp = JSApiMetaServerRemoveResponse{ApiResponse: ApiResponse{Type: JSApiMetaServerRemoveResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	if isEmptyRequest(msg) {
 		resp.Error = NewJSBadRequestError()
@@ -2656,7 +2732,7 @@ func (s *Server) jsLeaderServerStreamMoveRequest(sub *subscription, c *client, _
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -2684,6 +2760,11 @@ func (s *Server) jsLeaderServerStreamMoveRequest(sub *subscription, c *client, _
 	}
 
 	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	var req JSApiMetaServerStreamMoveRequest
 	if err := s.unmarshalRequest(c, acc, subject, msg, &req); err != nil {
@@ -2815,7 +2896,7 @@ func (s *Server) jsLeaderServerStreamCancelMoveRequest(sub *subscription, c *cli
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -2836,6 +2917,11 @@ func (s *Server) jsLeaderServerStreamCancelMoveRequest(sub *subscription, c *cli
 	}
 
 	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	accName := tokenAt(subject, 6)
 	streamName := tokenAt(subject, 7)
@@ -2925,7 +3011,7 @@ func (s *Server) jsLeaderAccountPurgeRequest(sub *subscription, c *client, _ *Ac
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -2942,6 +3028,11 @@ func (s *Server) jsLeaderAccountPurgeRequest(sub *subscription, c *client, _ *Ac
 	accName := tokenAt(subject, 5)
 
 	var resp = JSApiAccountPurgeResponse{ApiResponse: ApiResponse{Type: JSApiAccountPurgeResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	if !s.JetStreamIsClustered() {
 		var streams []*stream
@@ -3012,7 +3103,7 @@ func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, _ *Accoun
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -3040,6 +3131,11 @@ func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, _ *Accoun
 
 	var preferredLeader string
 	var resp = JSApiLeaderStepDownResponse{ApiResponse: ApiResponse{Type: JSApiLeaderStepDownResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	if isJSONObjectOrArray(msg) {
 		var req JSApiLeaderStepdownRequest
@@ -3185,13 +3281,18 @@ func (s *Server) jsStreamDeleteRequest(sub *subscription, c *client, _ *Account,
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamDeleteResponse{ApiResponse: ApiResponse{Type: JSApiStreamDeleteResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -3253,7 +3354,7 @@ func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, _ *Account, su
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -3262,6 +3363,11 @@ func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, _ *Account, su
 	stream := tokenAt(subject, 6)
 
 	var resp = JSApiMsgDeleteResponse{ApiResponse: ApiResponse{Type: JSApiMsgDeleteResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// If we are in clustered mode we need to be the stream leader to proceed.
 	if s.JetStreamIsClustered() {
@@ -3372,7 +3478,7 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, _ *Account, subje
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -3381,6 +3487,11 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, _ *Account, subje
 	stream := tokenAt(subject, 6)
 
 	var resp = JSApiMsgGetResponse{ApiResponse: ApiResponse{Type: JSApiMsgGetResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// If we are in clustered mode we need to be the stream leader to proceed.
 	if s.JetStreamIsClustered() {
@@ -3526,7 +3637,7 @@ func (s *Server) jsConsumerUnpinRequest(sub *subscription, c *client, _ *Account
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -3537,6 +3648,11 @@ func (s *Server) jsConsumerUnpinRequest(sub *subscription, c *client, _ *Account
 
 	var req JSApiConsumerUnpinRequest
 	var resp = JSApiConsumerUnpinResponse{ApiResponse: ApiResponse{Type: JSApiConsumerUnpinResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	if err := json.Unmarshal(msg, &req); err != nil {
 		resp.Error = NewJSInvalidJSONError(err)
@@ -3642,7 +3758,7 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, _ *Account,
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -3651,6 +3767,11 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, _ *Account,
 	stream := streamNameFromSubject(subject)
 
 	var resp = JSApiStreamPurgeResponse{ApiResponse: ApiResponse{Type: JSApiStreamPurgeResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// If we are in clustered mode we need to be the stream leader to proceed.
 	if s.JetStreamIsClustered() {
@@ -3784,13 +3905,18 @@ func (s *Server) jsStreamRestoreRequest(sub *subscription, c *client, _ *Account
 	if c == nil || !s.JetStreamIsLeader() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiStreamRestoreResponse{ApiResponse: ApiResponse{Type: JSApiStreamRestoreResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if !acc.JetStreamEnabled() {
 		resp.Error = NewJSNotEnabledForAccountError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -4064,7 +4190,7 @@ func (s *Server) jsStreamSnapshotRequest(sub *subscription, c *client, _ *Accoun
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -4079,6 +4205,11 @@ func (s *Server) jsStreamSnapshotRequest(sub *subscription, c *client, _ *Accoun
 	}
 
 	var resp = JSApiStreamSnapshotResponse{ApiResponse: ApiResponse{Type: JSApiStreamSnapshotResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if !acc.JetStreamEnabled() {
 		resp.Error = NewJSNotEnabledForAccountError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, smsg, s.jsonResponse(&resp))
@@ -4300,13 +4431,18 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	var req CreateConsumerRequest
 	if err := s.unmarshalRequest(c, acc, subject, msg, &req); err != nil {
@@ -4503,7 +4639,7 @@ func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, _ *Account
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -4513,6 +4649,11 @@ func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, _ *Account
 		ApiResponse: ApiResponse{Type: JSApiConsumerNamesResponseType},
 		Consumers:   []string{},
 	}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -4625,7 +4766,7 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -4635,6 +4776,11 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account,
 		ApiResponse: ApiResponse{Type: JSApiConsumerListResponseType},
 		Consumers:   []*ConsumerInfo{},
 	}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -4653,6 +4799,12 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account,
 		}
 	}
 
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSClusterNotAvailError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
 	if hasJS, doErr := acc.checkJetStream(); !hasJS {
 		if doErr {
 			resp.Error = NewJSNotEnabledForAccountError()
@@ -4718,7 +4870,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -4728,6 +4880,11 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
 	consumerName := consumerNameFromSubject(subject)
 
 	var resp = JSApiConsumerInfoResponse{ApiResponse: ApiResponse{Type: JSApiConsumerInfoResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	if !isEmptyRequest(msg) {
 		resp.Error = NewJSNotEmptyRequestError()
@@ -4901,13 +5058,18 @@ func (s *Server) jsConsumerDeleteRequest(sub *subscription, c *client, _ *Accoun
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
 	}
 
 	var resp = JSApiConsumerDeleteResponse{ApiResponse: ApiResponse{Type: JSApiConsumerDeleteResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	// Determine if we should proceed here when we are in clustered mode.
 	if s.JetStreamIsClustered() {
@@ -4973,7 +5135,7 @@ func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
-	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	ci, acc, hdr, msg, err := s.getRequestInfo(c, rmsg)
 	if err != nil {
 		s.Warnf(badAPIRequestT, msg)
 		return
@@ -4981,6 +5143,11 @@ func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account
 
 	var req JSApiConsumerPauseRequest
 	var resp = JSApiConsumerPauseResponse{ApiResponse: ApiResponse{Type: JSApiConsumerPauseResponseType}}
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 
 	if isJSONObjectOrArray(msg) {
 		if err := s.unmarshalRequest(c, acc, subject, msg, &req); err != nil {
diff --git a/server/jetstream_batching.go b/server/jetstream_batching.go
index 6b7fc01a..2c40ff27 100644
--- a/server/jetstream_batching.go
+++ b/server/jetstream_batching.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"math"
 	"math/big"
+	"path/filepath"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -41,26 +42,69 @@ type batchGroup struct {
 	timer *time.Timer
 }
 
-func newBatchGroup(s *Server, batchId string, batches *batching, store StreamStore) *batchGroup {
+// Lock should be held.
+func (batches *batching) newBatchGroup(mset *stream, batchId string) (*batchGroup, error) {
+	store, err := newBatchStore(mset, batchId)
+	if err != nil {
+		return nil, err
+	}
 	b := &batchGroup{store: store}
 
 	// Create a timer to clean up after timeout.
 	timeout := streamMaxBatchTimeout
-	if maxBatchTimeout := s.getOpts().JetStreamLimits.MaxBatchTimeout; maxBatchTimeout > 0 {
+	if maxBatchTimeout := mset.srv.getOpts().JetStreamLimits.MaxBatchTimeout; maxBatchTimeout > 0 {
 		timeout = maxBatchTimeout
 	}
 	b.timer = time.AfterFunc(timeout, func() {
 		b.cleanup(batchId, batches)
 	})
-	return b
+	return b, nil
+}
+
+func getBatchStoreDir(mset *stream, batchId string) (string, string) {
+	mset.mu.RLock()
+	jsa, name := mset.jsa, mset.cfg.Name
+	mset.mu.RUnlock()
+
+	jsa.mu.RLock()
+	sd := jsa.storeDir
+	jsa.mu.RUnlock()
+
+	bname := getHash(batchId)
+	return bname, filepath.Join(sd, streamsDir, name, batchesDir, bname)
+}
+
+func newBatchStore(mset *stream, batchId string) (StreamStore, error) {
+	mset.mu.RLock()
+	replicas, storage := mset.cfg.Replicas, mset.cfg.Storage
+	mset.mu.RUnlock()
+
+	if replicas == 1 && storage == FileStorage {
+		bname, storeDir := getBatchStoreDir(mset, batchId)
+		fcfg := FileStoreConfig{AsyncFlush: true, BlockSize: defaultLargeBlockSize, StoreDir: storeDir}
+		s := mset.srv
+		prf := s.jsKeyGen(s.getOpts().JetStreamKey, mset.acc.Name)
+		if prf != nil {
+			// We are encrypted here, fill in correct cipher selection.
+			fcfg.Cipher = s.getOpts().JetStreamCipher
+		}
+		oldprf := s.jsKeyGen(s.getOpts().JetStreamOldKey, mset.acc.Name)
+		cfg := StreamConfig{Name: bname, Storage: FileStorage}
+		return newFileStoreWithCreated(fcfg, cfg, time.Time{}, prf, oldprf)
+	}
+	return newMemStore(&StreamConfig{Name: _EMPTY_, Storage: MemoryStorage})
 }
 
-// stopCleanupTimerLocked indicates the batch is ready to be committed.
+// readyForCommit indicates the batch is ready to be committed.
 // If the timer has already cleaned up the batch, we can't commit.
 // Otherwise, we ensure the timer does not clean up the batch in the meantime.
 // Lock should be held.
-func (b *batchGroup) stopCleanupTimerLocked() bool {
-	return b.timer.Stop()
+func (b *batchGroup) readyForCommit() bool {
+	if !b.timer.Stop() {
+		return false
+	}
+	b.store.FlushAllPending()
+	return true
 }
 
 // cleanup deletes underlying resources associated with the batch and unregisters it from the stream's batches.
@@ -74,10 +118,17 @@ func (b *batchGroup) cleanup(batchId string, batches *batching) {
 func (b *batchGroup) cleanupLocked(batchId string, batches *batching) {
 	globalInflightBatches.Add(-1)
 	b.timer.Stop()
-	b.store.Delete()
+	b.store.Delete(true)
 	delete(batches.group, batchId)
 }
 
+// Lock should be held.
+func (b *batchGroup) stopLocked() {
+	globalInflightBatches.Add(-1)
+	b.timer.Stop()
+	b.store.Stop()
+}
+
 // batchStagedDiff stages all changes for consistency checks until commit.
 type batchStagedDiff struct {
 	msgIds             map[string]struct{}
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 21ee0868..d765f546 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -7429,6 +7429,7 @@ func (cc *jetStreamCluster) createGroupForConsumer(cfg *ConsumerConfig, sa *stre
 		return nil
 	}
 
+	replicas := cfg.replicas(sa.Config)
 	peers := copyStrings(sa.Group.Peers)
 	var _ss [5]string
 	active := _ss[:0]
@@ -7441,20 +7442,20 @@ func (cc *jetStreamCluster) createGroupForConsumer(cfg *ConsumerConfig, sa *stre
 			}
 		}
 	}
-	if quorum := cfg.Replicas/2 + 1; quorum > len(active) {
+	if quorum := replicas/2 + 1; quorum > len(active) {
 		// Not enough active to satisfy the request.
 		return nil
 	}
 
 	// If we want less then our parent stream, select from active.
-	if cfg.Replicas > 0 && cfg.Replicas < len(peers) {
+	if replicas > 0 && replicas < len(peers) {
 		// Pedantic in case stream is say R5 and consumer is R3 and 3 or more offline, etc.
-		if len(active) < cfg.Replicas {
+		if len(active) < replicas {
 			return nil
 		}
 		// First shuffle the active peers and then select to account for replica = 1.
 		rand.Shuffle(len(active), func(i, j int) { active[i], active[j] = active[j], active[i] })
-		peers = active[:cfg.Replicas]
+		peers = active[:replicas]
 	}
 	storage := sa.Config.Storage
 	if cfg.MemoryStorage {
@@ -7640,12 +7641,6 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 
 		// We need to set the ephemeral here before replicating.
 		if !isDurableConsumer(cfg) {
-			// We chose to have ephemerals be R=1 unless stream is interest or workqueue.
-			// Consumer can override.
-			if sa.Config.Retention == LimitsPolicy && cfg.Replicas <= 1 {
-				rg.Peers = []string{rg.Preferred}
-				rg.Name = groupNameForConsumer(rg.Peers, rg.Storage)
-			}
 			if cfg.Name != _EMPTY_ {
 				oname = cfg.Name
 			} else {
@@ -8065,7 +8060,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	discard, discardNewPer, maxMsgs, maxMsgsPer, maxBytes := mset.cfg.Discard, mset.cfg.DiscardNewPer, mset.cfg.MaxMsgs, mset.cfg.MaxMsgsPer, mset.cfg.MaxBytes
 	s, js, jsa, st, r, tierName, outq, node := mset.srv, mset.js, mset.jsa, mset.cfg.Storage, mset.cfg.Replicas, mset.tier, mset.outq, mset.node
 	maxMsgSize, lseq := int(mset.cfg.MaxMsgSize), mset.lseq
-	isLeader, isSealed, allowRollup, denyPurge, allowTTL, allowMsgCounter, allowAtomicPublish := mset.isLeader(), mset.cfg.Sealed, mset.cfg.AllowRollup, mset.cfg.DenyPurge, mset.cfg.AllowMsgTTL, mset.cfg.AllowMsgCounter, mset.cfg.AllowAtomicPublish
+	isLeader, isSealed, allowRollup, denyPurge, allowTTL, allowMsgCounter := mset.isLeader(), mset.cfg.Sealed, mset.cfg.AllowRollup, mset.cfg.DenyPurge, mset.cfg.AllowMsgTTL, mset.cfg.AllowMsgCounter
 	mset.mu.RUnlock()
 
 	// This should not happen but possible now that we allow scale up, and scale down where this could trigger.
@@ -8144,244 +8139,6 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		return err
 	}
 
-	if batchId := getBatchId(hdr); batchId != _EMPTY_ {
-		if !allowAtomicPublish {
-			err := NewJSAtomicPublishDisabledError()
-			if canRespond {
-				b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
-				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
-			}
-			return err
-		}
-
-		// Batch ID is too long.
-		if len(batchId) > 64 {
-			err := NewJSAtomicPublishInvalidBatchIDError()
-			if canRespond {
-				b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
-				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
-			}
-			return err
-		}
-
-		batchSeq, exists := getBatchSequence(hdr)
-		if !exists {
-			err := NewJSAtomicPublishMissingSeqError()
-			if canRespond {
-				b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
-				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
-			}
-			return err
-		}
-		commit := len(sliceHeader(JSBatchCommit, hdr)) != 0
-
-		mset.mu.Lock()
-		if mset.batches == nil {
-			mset.batches = &batching{
-				group: make(map[string]*batchGroup, 1),
-			}
-		}
-		batches := mset.batches
-		mset.mu.Unlock()
-
-		respondIncompleteBatch := func() error {
-			err := NewJSAtomicPublishIncompleteBatchError()
-			if canRespond {
-				buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
-				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
-			}
-			return err
-		}
-
-		// Get batch.
-		batches.mu.Lock()
-		b, ok := batches.group[batchId]
-		if !ok {
-			if batchSeq != 1 {
-				batches.mu.Unlock()
-				return respondIncompleteBatch()
-			}
-
-			// Limits.
-			maxInflightPerStream := streamMaxBatchInflightPerStream
-			maxInflightTotal := streamMaxBatchInflightTotal
-			opts := s.getOpts()
-			if opts.JetStreamLimits.MaxBatchInflightPerStream > 0 {
-				maxInflightPerStream = opts.JetStreamLimits.MaxBatchInflightPerStream
-			}
-			if opts.JetStreamLimits.MaxBatchInflightTotal > 0 {
-				maxInflightTotal = opts.JetStreamLimits.MaxBatchInflightTotal
-			}
-
-			// Confirm we can facilitate an additional batch.
-			if len(batches.group)+1 > maxInflightPerStream {
-				batches.mu.Unlock()
-				return respondIncompleteBatch()
-			}
-
-			// Confirm we'll not exceed the server limit.
-			if globalInflightBatches.Add(1) > int32(maxInflightTotal) {
-				globalInflightBatches.Add(-1)
-				batches.mu.Unlock()
-				return respondIncompleteBatch()
-			}
-
-			ms, err := newMemStore(&StreamConfig{Name: _EMPTY_, Storage: MemoryStorage})
-			if err != nil {
-				globalInflightBatches.Add(-1)
-				batches.mu.Unlock()
-				return respondIncompleteBatch()
-			}
-
-			b = newBatchGroup(s, batchId, batches, ms)
-			batches.group[batchId] = b
-		}
-
-		// Detect gaps.
-		b.lseq++
-		if b.lseq != batchSeq {
-			b.cleanupLocked(batchId, batches)
-			batches.mu.Unlock()
-			return respondIncompleteBatch()
-		}
-
-		// Confirm the batch doesn't exceed the allowed size.
-		maxSize := streamMaxBatchSize
-		if maxBatchSize := s.getOpts().JetStreamLimits.MaxBatchSize; maxBatchSize > 0 {
-			maxSize = maxBatchSize
-		}
-		if batchSeq > uint64(maxSize) {
-			b.cleanupLocked(batchId, batches)
-			batches.mu.Unlock()
-			return respondIncompleteBatch()
-		}
-
-		// Persist, but optimize if we're committing because we already know last.
-		if !commit {
-			seq, _, err := b.store.StoreMsg(subject, hdr, msg, 0)
-			if err != nil || seq != batchSeq {
-				b.cleanupLocked(batchId, batches)
-				batches.mu.Unlock()
-				return respondIncompleteBatch()
-			}
-			batches.mu.Unlock()
-			return nil
-		}
-
-		// Ensure the batch is prepared for the commit and will not be cleaned up while committing.
-		if !b.stopCleanupTimerLocked() {
-			// Don't do cleanup, this is already done.
-			batches.mu.Unlock()
-			return respondIncompleteBatch()
-		}
-
-		// Proceed with proposing the batch.
-
-		// We only use mset.clseq for clustering and in case we run ahead of actual commits.
-		// Check if we need to set initial value here
-		mset.clMu.Lock()
-		if mset.clseq == 0 || mset.clseq < lseq+mset.clfs {
-			// Re-capture
-			lseq = mset.lastSeq()
-			mset.clseq = lseq + mset.clfs
-		}
-
-		// TODO(mvv): support message tracing
-		ts := time.Now().UnixNano()
-		var entries []*Entry
-
-		var (
-			bsubj  string
-			bhdr   []byte
-			bmsg   []byte
-			apiErr *ApiError
-			err    error
-			smv    StoreMsg
-			sm     *StoreMsg
-			sz     int
-		)
-
-		errorOnUnsupported := func(seq uint64, header string) *ApiError {
-			apiErr := NewJSAtomicPublishUnsupportedHeaderBatchError(header)
-			// TODO(mvv): reset in-memory expected header maps
-			mset.clseq -= seq - 1
-			mset.clMu.Unlock()
-			b.cleanupLocked(batchId, batches)
-			batches.mu.Unlock()
-			if canRespond {
-				buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: apiErr})
-				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
-			}
-			return apiErr
-		}
-
-		diff := &batchStagedDiff{}
-		for seq := uint64(1); seq <= batchSeq; seq++ {
-			if seq == batchSeq {
-				bsubj, bhdr, bmsg = subject, hdr, msg
-			} else if sm, err = b.store.LoadMsg(seq, &smv); sm != nil && err == nil {
-				bsubj, bhdr, bmsg = sm.subj, sm.hdr, sm.msg
-			} else {
-				mset.clseq -= seq - 1
-				mset.clMu.Unlock()
-				b.cleanupLocked(batchId, batches)
-				batches.mu.Unlock()
-				return respondIncompleteBatch()
-			}
-
-			// Reject unsupported headers.
-			if msgId := getMsgId(bhdr); msgId != _EMPTY_ {
-				return errorOnUnsupported(seq, JSMsgId)
-			}
-			if getExpectedLastMsgId(hdr) != _EMPTY_ {
-				return errorOnUnsupported(seq, JSExpectedLastMsgId)
-			}
-
-			if bhdr, bmsg, _, apiErr, err = checkMsgHeadersPreClusteredProposal(diff, mset, subject, bhdr, bmsg, sourced, name, jsa, allowRollup, denyPurge, allowTTL, allowMsgCounter, discard, discardNewPer, maxMsgSize, maxMsgs, maxMsgsPer, maxBytes); err != nil {
-				// TODO(mvv): reset in-memory expected header maps
-				mset.clseq -= seq - 1
-				mset.clMu.Unlock()
-				b.cleanupLocked(batchId, batches)
-				batches.mu.Unlock()
-				if canRespond {
-					buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: apiErr})
-					outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
-				}
-				return err
-			}
-
-			var _reply string
-			if seq == batchSeq {
-				_reply = reply
-			}
-			esm := encodeStreamMsgAllowCompress(bsubj, _reply, bhdr, bmsg, mset.clseq, ts, sourced)
-			entries = append(entries, newEntry(EntryNormal, esm))
-			mset.clseq++
-			sz += len(esm)
-		}
-
-		// Do proposal.
-		diff.commit(mset)
-		// TODO(mvv): replace with individual `node.Propose`?
-		if err := node.ProposeMulti(entries); err == nil {
-			mset.trackReplicationTraffic(node, sz, r)
-		} else {
-			// TODO(mvv): reset in-memory expected header maps
-			mset.clseq -= batchSeq
-		}
-
-		// Check to see if we are being overrun.
-		// TODO(dlc) - Make this a limit where we drop messages to protect ourselves, but allow to be configured.
-		if mset.clseq-(lseq+mset.clfs) > streamLagWarnThreshold {
-			lerr := fmt.Errorf("JetStream stream '%s > %s' has high message lag", jsa.acc().Name, name)
-			s.RateLimitWarnf("%s", lerr.Error())
-		}
-		mset.clMu.Unlock()
-		b.cleanupLocked(batchId, batches)
-		batches.mu.Unlock()
-		return nil
-	}
-
 	// Proceed with proposing this message.
 
 	// We only use mset.clseq for clustering and in case we run ahead of actual commits.
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 12055ece..660722dd 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -362,6 +362,9 @@ const (
 	// JSReplicasCountCannotBeNegative replicas count cannot be negative
 	JSReplicasCountCannotBeNegative ErrorIdentifier = 10133
 
+	// JSRequiredApiLevelErr JetStream minimum api level required
+	JSRequiredApiLevelErr ErrorIdentifier = 10185
+
 	// JSRestoreSubscribeFailedErrF JetStream unable to subscribe to restore snapshot {subject}: {err}
 	JSRestoreSubscribeFailedErrF ErrorIdentifier = 10042
 
@@ -676,6 +679,7 @@ var (
 		JSPeerRemapErr:                             {Code: 503, ErrCode: 10075, Description: "peer remap failed"},
 		JSRaftGeneralErrF:                          {Code: 500, ErrCode: 10041, Description: "{err}"},
 		JSReplicasCountCannotBeNegative:            {Code: 400, ErrCode: 10133, Description: "replicas count cannot be negative"},
+		JSRequiredApiLevelErr:                      {Code: 412, ErrCode: 10185, Description: "JetStream minimum api level required"},
 		JSRestoreSubscribeFailedErrF:               {Code: 500, ErrCode: 10042, Description: "JetStream unable to subscribe to restore snapshot {subject}: {err}"},
 		JSSequenceNotFoundErrF:                     {Code: 400, ErrCode: 10043, Description: "sequence {seq} not found"},
 		JSSnapshotDeliverSubjectInvalidErr:         {Code: 400, ErrCode: 10015, Description: "deliver subject not valid"},
@@ -2063,6 +2067,16 @@ func NewJSReplicasCountCannotBeNegativeError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSReplicasCountCannotBeNegative]
 }
 
+// NewJSRequiredApiLevelError creates a new JSRequiredApiLevelErr error: "JetStream minimum api level required"
+func NewJSRequiredApiLevelError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSRequiredApiLevelErr]
+}
+
 // NewJSRestoreSubscribeFailedError creates a new JSRestoreSubscribeFailedErrF error: "JetStream unable to subscribe to restore snapshot {subject}: {err}"
 func NewJSRestoreSubscribeFailedError(err error, subject interface{}, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/jetstream_versioning.go b/server/jetstream_versioning.go
index e62db0fc..ebdfcee6 100644
--- a/server/jetstream_versioning.go
+++ b/server/jetstream_versioning.go
@@ -55,11 +55,6 @@ func setStaticStreamMetadata(cfg *StreamConfig) {
 		requires(2)
 	}
 
-	// Async flush was added in v2.12 and require API level 2.
-	if cfg.AllowAsyncFlush {
-		requires(2)
-	}
-
 	cfg.Metadata[JSRequiredLevelMetadataKey] = strconv.Itoa(requiredApiLevel)
 }
 
@@ -131,11 +126,6 @@ func setStaticConsumerMetadata(cfg *ConsumerConfig) {
 		requires(1)
 	}
 
-	// Added in 2.12, absent | zero is the feature is not used.
-	if cfg.MinLastSeq > 0 {
-		requires(2)
-	}
-
 	cfg.Metadata[JSRequiredLevelMetadataKey] = strconv.Itoa(requiredApiLevel)
 }
 
@@ -197,3 +187,13 @@ func deleteDynamicMetadata(metadata map[string]string) {
 	delete(metadata, JSServerVersionMetadataKey)
 	delete(metadata, JSServerLevelMetadataKey)
 }
+
+// errorOnRequiredApiLevel returns whether a request should be rejected based on the JSRequiredApiLevel header.
+func errorOnRequiredApiLevel(hdr []byte) bool {
+	reqApiLevel := sliceHeader(JSRequiredApiLevel, hdr)
+	if len(reqApiLevel) == 0 {
+		return false
+	}
+	minLevel, err := strconv.Atoi(string(reqApiLevel))
+	return err != nil || JSApiLevel < minLevel
+}
diff --git a/server/leafnode.go b/server/leafnode.go
index 8fa10f04..6122377b 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -756,6 +756,7 @@ func (s *Server) startLeafNodeAcceptLoop() {
 		Domain:        opts.JetStreamDomain,
 		Proto:         s.getServerProto(),
 		InfoOnConnect: true,
+		JSApiLevel:    JSApiLevel,
 	}
 	// If we have selected a random port...
 	if port == 0 {
@@ -1065,6 +1066,8 @@ func (s *Server) createLeafNode(conn net.Conn, rURL *url.URL, remote *leafNodeCf
 		if cm := opts.LeafNode.Compression.Mode; cm != CompressionNotSupported {
 			info.Compression = cm
 		}
+		// We always send a nonce for LEAF connections. Do not change that without
+		// taking into account presence of proxy trusted keys.
 		s.generateNonce(nonce[:])
 		s.mu.Unlock()
 	}
@@ -1804,9 +1807,13 @@ func (s *Server) removeLeafNodeConnection(c *client) {
 			c.leaf.gwSub = nil
 		}
 	}
+	proxyKey := c.proxyKey
 	c.mu.Unlock()
 	s.mu.Lock()
 	delete(s.leafs, cid)
+	if proxyKey != _EMPTY_ {
+		s.removeProxiedConn(proxyKey, cid)
+	}
 	s.mu.Unlock()
 	s.removeFromTempClients(cid)
 }
diff --git a/server/memstore.go b/server/memstore.go
index 639502bc..4b550085 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1966,7 +1966,7 @@ func memStoreMsgSize(subj string, hdr, msg []byte) uint64 {
 }
 
 // Delete is same as Stop for memory store.
-func (ms *memStore) Delete() error {
+func (ms *memStore) Delete(_ bool) error {
 	return ms.Stop()
 }
 
diff --git a/server/monitor.go b/server/monitor.go
index c799256e..9eb2ef2d 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -146,11 +146,17 @@ type ConnInfo struct {
 	NameTag        string         `json:"name_tag,omitempty"`
 	Tags           jwt.TagList    `json:"tags,omitempty"`
 	MQTTClient     string         `json:"mqtt_client,omitempty"` // This is the MQTT client id
+	Proxy          *ProxyInfo     `json:"proxy,omitempty"`
 
 	// Internal
 	rtt int64 // For fast sorting
 }
 
+// ProxyInfo represents the information about this proxied connection.
+type ProxyInfo struct {
+	Key string `json:"key"`
+}
+
 // TLSPeerCert contains basic information about a TLS peer certificate
 type TLSPeerCert struct {
 	Subject          string `json:"subject,omitempty"`
@@ -571,6 +577,7 @@ func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time, auth bool)
 	// we need to use atomic here.
 	ci.InMsgs = atomic.LoadInt64(&client.inMsgs)
 	ci.InBytes = atomic.LoadInt64(&client.inBytes)
+	ci.Proxy = createProxyInfo(client)
 
 	// If the connection is gone, too bad, we won't set TLSVersion and TLSCipher.
 	// Exclude clients that are still doing handshake so we don't block in
@@ -593,6 +600,17 @@ func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time, auth bool)
 	}
 }
 
+// If this client came from a trusted proxy, this will return a ProxyInfo
+// to be used in ConnInfo or LeafInfo.
+//
+// Client lock must be held on entry.
+func createProxyInfo(c *client) *ProxyInfo {
+	if c.proxyKey == _EMPTY_ {
+		return nil
+	}
+	return &ProxyInfo{Key: c.proxyKey}
+}
+
 func makePeerCerts(pc []*x509.Certificate) []*TLSPeerCert {
 	res := make([]*TLSPeerCert, len(pc))
 	for i, c := range pc {
@@ -1254,6 +1272,7 @@ type Varz struct {
 	PinnedAccountFail     uint64                 `json:"pinned_account_fails,omitempty"`    // PinnedAccountFail is how often user logon fails due to the issuer account not being pinned.
 	OCSPResponseCache     *OCSPResponseCacheVarz `json:"ocsp_peer_cache,omitempty"`         // OCSPResponseCache is the state of the OCSP cache // OCSPResponseCache holds information about
 	SlowConsumersStats    *SlowConsumersStats    `json:"slow_consumer_stats"`               // SlowConsumersStats is statistics about all detected Slow Consumer
+	Proxies               *ProxiesOptsVarz       `json:"proxies,omitempty"`
 }
 
 // JetStreamVarz contains basic runtime information about jetstream
@@ -1370,6 +1389,16 @@ type OCSPResponseCacheVarz struct {
 	Unknowns  int64  `json:"cached_unknown_responses,omitempty"` // Unknowns  is how many of the stored cache entries are unknown responses
 }
 
+// ProxiesOptsVarz contains monitoring proxies information
+type ProxiesOptsVarz struct {
+	Trusted []*ProxyOptsVarz `json:"trusted,omitempty"`
+}
+
+// ProxyOptsVarz contains monitoring proxy information
+type ProxyOptsVarz struct {
+	Key string `json:"key"`
+}
+
 // VarzOptions are the options passed to Varz().
 // Currently, there are no options defined.
 type VarzOptions struct{}
@@ -1720,6 +1749,19 @@ func (s *Server) updateVarzConfigReloadableFields(v *Varz) {
 	v.Websocket.TLSPinnedCerts = getPinnedCertsAsSlice(opts.Websocket.TLSPinnedCerts)
 
 	v.TLSOCSPPeerVerify = s.ocspPeerVerify && v.TLSRequired && s.opts.tlsConfigOpts != nil && s.opts.tlsConfigOpts.OCSPPeerConfig != nil && s.opts.tlsConfigOpts.OCSPPeerConfig.Verify
+
+	if opts.Proxies != nil {
+		if v.Proxies == nil {
+			v.Proxies = &ProxiesOptsVarz{}
+		}
+		trusted := make([]*ProxyOptsVarz, 0, len(opts.Proxies.Trusted))
+		for _, t := range opts.Proxies.Trusted {
+			trusted = append(trusted, &ProxyOptsVarz{Key: t.Key})
+		}
+		v.Proxies.Trusted = trusted
+	} else {
+		v.Proxies = nil
+	}
 }
 
 func getPinnedCertsAsSlice(certs PinnedCertSet) []string {
@@ -2254,20 +2296,21 @@ type LeafzOptions struct {
 
 // LeafInfo has detailed information on each remote leafnode connection.
 type LeafInfo struct {
-	ID          uint64   `json:"id"`
-	Name        string   `json:"name"`
-	IsSpoke     bool     `json:"is_spoke"`
-	Account     string   `json:"account"`
-	IP          string   `json:"ip"`
-	Port        int      `json:"port"`
-	RTT         string   `json:"rtt,omitempty"`
-	InMsgs      int64    `json:"in_msgs"`
-	OutMsgs     int64    `json:"out_msgs"`
-	InBytes     int64    `json:"in_bytes"`
-	OutBytes    int64    `json:"out_bytes"`
-	NumSubs     uint32   `json:"subscriptions"`
-	Subs        []string `json:"subscriptions_list,omitempty"`
-	Compression string   `json:"compression,omitempty"`
+	ID          uint64     `json:"id"`
+	Name        string     `json:"name"`
+	IsSpoke     bool       `json:"is_spoke"`
+	Account     string     `json:"account"`
+	IP          string     `json:"ip"`
+	Port        int        `json:"port"`
+	RTT         string     `json:"rtt,omitempty"`
+	InMsgs      int64      `json:"in_msgs"`
+	OutMsgs     int64      `json:"out_msgs"`
+	InBytes     int64      `json:"in_bytes"`
+	OutBytes    int64      `json:"out_bytes"`
+	NumSubs     uint32     `json:"subscriptions"`
+	Subs        []string   `json:"subscriptions_list,omitempty"`
+	Compression string     `json:"compression,omitempty"`
+	Proxy       *ProxyInfo `json:"proxy,omitempty"`
 }
 
 // Leafz returns a Leafz structure containing information about leafnodes.
@@ -2310,6 +2353,7 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 				OutBytes:    ln.outBytes,
 				NumSubs:     uint32(len(ln.subs)),
 				Compression: ln.leaf.compression,
+				Proxy:       createProxyInfo(ln),
 			}
 			if opts != nil && opts.Subscriptions {
 				lni.Subs = make([]string, 0, len(ln.subs))
@@ -2522,6 +2566,10 @@ func (reason ClosedState) String() string {
 		return "Cluster Names Identical"
 	case Kicked:
 		return "Kicked"
+	case ProxyNotTrusted:
+		return "Proxy Not Trusted"
+	case ProxyRequired:
+		return "Proxy Required"
 	}
 
 	return "Unknown State"
diff --git a/server/nkey.go b/server/nkey.go
index f0579177..97195b0e 100644
--- a/server/nkey.go
+++ b/server/nkey.go
@@ -34,7 +34,7 @@ func (s *Server) NonceRequired() bool {
 // nonceRequired tells us if we should send a nonce.
 // Lock should be held on entry.
 func (s *Server) nonceRequired() bool {
-	return s.getOpts().AlwaysEnableNonce || len(s.nkeys) > 0 || s.trustedKeys != nil
+	return s.getOpts().AlwaysEnableNonce || len(s.nkeys) > 0 || s.trustedKeys != nil || len(s.proxiesKeyPairs) > 0
 }
 
 // Generate a nonce for INFO challenge.
diff --git a/server/opts.go b/server/opts.go
index 7a8d63ae..4cdf4bdc 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -151,6 +151,7 @@ type LeafNodeOpts struct {
 	Port           int           `json:"port,omitempty"`
 	Username       string        `json:"-"`
 	Password       string        `json:"-"`
+	ProxyRequired  bool          `json:"-"`
 	Nkey           string        `json:"-"`
 	Account        string        `json:"-"`
 	Users          []*User       `json:"-"`
@@ -324,6 +325,7 @@ type Options struct {
 	NoSystemAccount            bool          `json:"-"`
 	Username                   string        `json:"-"`
 	Password                   string        `json:"-"`
+	ProxyRequired              bool          `json:"-"`
 	Authorization              string        `json:"-"`
 	AuthCallout                *AuthCallout  `json:"-"`
 	PingInterval               time.Duration `json:"ping_interval"`
@@ -445,6 +447,9 @@ type Options struct {
 	OCSPConfig    *OCSPConfig
 	tlsConfigOpts *TLSConfigOpts
 
+	// Proxies configuration.
+	Proxies *ProxiesConfig
+
 	// private fields, used to know if bool options are explicitly
 	// defined in config and/or command line params.
 	inConfig  map[string]bool
@@ -729,6 +734,8 @@ type authorization struct {
 	token string
 	nkey  string
 	acc   string
+	// If connection must come through proxy
+	proxyRequired bool
 	// Multiple Nkeys/Users
 	nkeys              []*NkeyUser
 	users              []*User
@@ -781,6 +788,17 @@ type OCSPConfig struct {
 	OverrideURLs []string
 }
 
+// ProxiesConfig represents the options of Proxies.
+type ProxiesConfig struct {
+	Trusted []*ProxyConfig
+}
+
+// ProxyConfig represents the options of Proxy.
+type ProxyConfig struct {
+	// Public key.
+	Key string
+}
+
 var tlsUsage = `
 TLS configuration is specified in the tls section of a configuration file:
 
@@ -1063,6 +1081,7 @@ func (o *Options) processConfigFileLine(k string, v any, errors *[]error, warnin
 		o.authBlockDefined = true
 		o.Username = auth.user
 		o.Password = auth.pass
+		o.ProxyRequired = auth.proxyRequired
 		o.Authorization = auth.token
 		o.AuthTimeout = auth.timeout
 		o.AuthCallout = auth.callout
@@ -1727,6 +1746,13 @@ func (o *Options) processConfigFileLine(k string, v any, errors *[]error, warnin
 		o.NoFastProducerStall = v.(bool)
 	case "max_closed_clients":
 		o.MaxClosedClients = int(v.(int64))
+	case "proxies":
+		proxies, err := parseProxies(tk, errors)
+		if err != nil {
+			*errors = append(*errors, err)
+			return
+		}
+		o.Proxies = proxies
 	default:
 		if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
 			err := &unknownConfigFieldErr{
@@ -2600,6 +2626,7 @@ func parseLeafNodes(v any, opts *Options, errors *[]error, warnings *[]error) er
 			}
 			opts.LeafNode.Username = auth.user
 			opts.LeafNode.Password = auth.pass
+			opts.LeafNode.ProxyRequired = auth.proxyRequired
 			opts.LeafNode.AuthTimeout = auth.timeout
 			opts.LeafNode.Account = auth.acc
 			opts.LeafNode.Users = auth.users
@@ -2726,6 +2753,8 @@ func parseLeafAuthorization(v any, errors, warnings *[]error) (*authorization, e
 			auth.users = users
 		case "account":
 			auth.acc = mv.(string)
+		case "proxy_required":
+			auth.proxyRequired = mv.(bool)
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
@@ -2784,6 +2813,8 @@ func parseLeafUsers(mv any, errors *[]error) ([]*User, error) {
 				// we need to create internal objects to store u/p and account
 				// name and have a server structure to hold that.
 				user.Account = NewAccount(v.(string))
+			case "proxy_required":
+				user.ProxyRequired = v.(bool)
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{
@@ -4273,6 +4304,8 @@ func parseAuthorization(v any, errors, warnings *[]error) (*authorization, error
 				continue
 			}
 			auth.callout = ac
+		case "proxy_required":
+			auth.proxyRequired = mv.(bool)
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
@@ -4345,6 +4378,9 @@ func parseUsers(mv any, errors *[]error) ([]*NkeyUser, []*User, error) {
 				cts := parseAllowedConnectionTypes(tk, &lt, v, errors)
 				nkey.AllowedConnectionTypes = cts
 				user.AllowedConnectionTypes = cts
+			case "proxy_required":
+				nkey.ProxyRequired = v.(bool)
+				user.ProxyRequired = v.(bool)
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{
@@ -5358,6 +5394,95 @@ func parseMQTT(v any, o *Options, errors *[]error, warnings *[]error) error {
 	return nil
 }
 
+func parseProxies(mv any, errors *[]error) (*ProxiesConfig, error) {
+	var (
+		tk      token
+		lt      token
+		proxies = &ProxiesConfig{}
+	)
+	defer convertPanicToErrorList(&lt, errors)
+
+	tk, mv = unwrapValue(mv, &lt)
+	pm, ok := mv.(map[string]any)
+	if !ok {
+		return nil, &configErr{tk, fmt.Sprintf("expected proxies to be a map/struct, got %T", mv)}
+	}
+	for mk, mv := range pm {
+		tk, _ = unwrapValue(mv, &lt)
+
+		switch strings.ToLower(mk) {
+		case "trusted":
+			trusted, err := parseProxiesTrusted(tk, errors)
+			if err != nil {
+				*errors = append(*errors, err)
+				continue
+			}
+			proxies.Trusted = trusted
+		default:
+			if !tk.IsUsedVariable() {
+				err := &unknownConfigFieldErr{
+					field: mk,
+					configErr: configErr{
+						token: tk,
+					},
+				}
+				*errors = append(*errors, err)
+			}
+		}
+	}
+	return proxies, nil
+}
+
+func parseProxiesTrusted(mv any, errors *[]error) ([]*ProxyConfig, error) {
+	var (
+		tk      token
+		lt      token
+		trusted []*ProxyConfig
+	)
+	defer convertPanicToErrorList(&lt, errors)
+
+	tk, mv = unwrapValue(mv, &lt)
+	ta, ok := mv.([]any)
+	if !ok {
+		return nil, &configErr{tk, fmt.Sprintf("expected proxies' trusted field to be an array, got %T", mv)}
+	}
+	for _, t := range ta {
+		tk, t = unwrapValue(t, &lt)
+		// Check its a map/struct
+		tm, ok := t.(map[string]any)
+		if !ok {
+			err := &configErr{tk, fmt.Sprintf("expected proxies' trusted entry to be a map/struct, got %T", t)}
+			*errors = append(*errors, err)
+			continue
+		}
+		proxy := &ProxyConfig{}
+		for k, v := range tm {
+			tk, v = unwrapValue(v, &lt)
+			switch strings.ToLower(k) {
+			case "key", "public_key":
+				proxy.Key = v.(string)
+				if !nkeys.IsValidPublicKey(proxy.Key) {
+					*errors = append(*errors, &configErr{tk, fmt.Sprintf("invalid proxy key %q", proxy.Key)})
+					continue
+				}
+			default:
+				if !tk.IsUsedVariable() {
+					err := &unknownConfigFieldErr{
+						field: k,
+						configErr: configErr{
+							token: tk,
+						},
+					}
+					*errors = append(*errors, err)
+					continue
+				}
+			}
+		}
+		trusted = append(trusted, proxy)
+	}
+	return trusted, nil
+}
+
 // GenTLSConfig loads TLS related configuration parameters.
 func GenTLSConfig(tc *TLSConfigOpts) (*tls.Config, error) {
 	// Create the tls.Config from our options before including the certs.
diff --git a/server/pse/pse_solaris.go b/server/pse/pse_solaris.go
index 8e40d2ed..f812ba48 100644
--- a/server/pse/pse_solaris.go
+++ b/server/pse/pse_solaris.go
@@ -1,4 +1,4 @@
-// Copyright 2015-2018 The NATS Authors
+// Copyright 2015-2025 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
@@ -10,14 +10,36 @@
 // 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.
+//
+// Copied from pse_openbsd.go
+
+//go:build illumos || solaris
 
 package pse
 
-// This is a placeholder for now.
-func ProcUsage(pcpu *float64, rss, vss *int64) error {
-	*pcpu = 0.0
-	*rss = 0
-	*vss = 0
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+)
 
+// ProcUsage returns CPU usage
+func ProcUsage(pcpu *float64, rss, vss *int64) error {
+	pidStr := fmt.Sprintf("%d", os.Getpid())
+	out, err := exec.Command("ps", "-o", "pcpu,rss,vsz", "-p", pidStr).Output()
+	if err != nil {
+		*rss, *vss = -1, -1
+		return fmt.Errorf("ps call failed:%v", err)
+	}
+	lines := strings.Split(string(out), "\n")
+	if len(lines) < 2 {
+		*rss, *vss = -1, -1
+		return fmt.Errorf("no ps output")
+	}
+	output := lines[1]
+	fmt.Sscanf(output, "%f %d %d", pcpu, rss, vss)
+	*rss *= 1024 // 1k blocks, want bytes.
+	*vss *= 1024 // 1k blocks, want bytes.
 	return nil
 }
diff --git a/server/raft.go b/server/raft.go
index e3e4f622..9870b284 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -95,7 +95,7 @@ type WAL interface {
 	State() StreamState
 	FastState(*StreamState)
 	Stop() error
-	Delete() error
+	Delete(inline bool) error
 }
 
 type Peer struct {
@@ -498,12 +498,14 @@ func (s *Server) initRaftNode(accName string, cfg *RaftConfig, labels pprofLabel
 			ae, err := n.loadEntry(index)
 			if err != nil {
 				n.warn("Could not load %d from WAL [%+v]: %v", index, state, err)
-				truncateAndErr(index)
+				// Truncate to the previous correct entry.
+				truncateAndErr(index - 1)
 				break
 			}
 			if ae.pindex != index-1 {
 				n.warn("Corrupt WAL, will truncate")
-				truncateAndErr(index)
+				// Truncate to the previous correct entry.
+				truncateAndErr(index - 1)
 				break
 			}
 			n.processAppendEntry(ae, nil)
@@ -1821,7 +1823,7 @@ func (n *raft) Delete() {
 	defer n.Unlock()
 
 	if wal := n.wal; wal != nil {
-		wal.Delete()
+		wal.Delete(false)
 	}
 	os.RemoveAll(n.sd)
 	n.debug("Deleted")
diff --git a/server/reload.go b/server/reload.go
index d5422e1c..4a49026a 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -1052,6 +1052,38 @@ func (s *Server) recheckPinnedCerts(curOpts *Options, newOpts *Options) {
 	}
 }
 
+type proxiesReload struct {
+	noopOption
+	add []string
+	del []string
+}
+
+func (p *proxiesReload) Apply(s *Server) {
+	var clients []*client
+	s.mu.Lock()
+	for _, k := range p.del {
+		cc := s.proxiedConns[k]
+		delete(s.proxiedConns, k)
+		if len(cc) > 0 {
+			for _, c := range cc {
+				clients = append(clients, c)
+			}
+		}
+	}
+	s.processProxiesTrustedKeys()
+	s.mu.Unlock()
+	if len(p.del) > 0 {
+		for _, c := range clients {
+			c.setAuthError(ErrAuthProxyNotTrusted)
+			c.authViolation()
+		}
+		s.Noticef("Reloaded: proxies trusted keys %q were removed", p.add)
+	}
+	if len(p.add) > 0 {
+		s.Noticef("Reloaded: proxies trusted keys %q were added", p.add)
+	}
+}
+
 // Reload reads the current configuration file and calls out to ReloadOptions
 // to apply the changes. This returns an error if the server was not started
 // with a config file or an option which doesn't support hot-swapping was changed.
@@ -1227,7 +1259,7 @@ func imposeOrder(value any) error {
 		slices.Sort(value.AllowedOrigins)
 	case string, bool, uint8, uint16, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
-		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher, *OCSPResponseCacheConfig:
+		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher, *OCSPResponseCacheConfig, *ProxiesConfig:
 		// explicitly skipped types
 	case *AuthCallout:
 	case JSTpmOpts:
@@ -1722,6 +1754,12 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			continue
 		case "nofastproducerstall":
 			diffOpts = append(diffOpts, &noFastProdStallReload{noStall: newValue.(bool)})
+		case "proxies":
+			new := newValue.(*ProxiesConfig)
+			old := oldValue.(*ProxiesConfig)
+			if add, del := diffProxiesTrustedKeys(old.Trusted, new.Trusted); len(add) > 0 || len(del) > 0 {
+				diffOpts = append(diffOpts, &proxiesReload{add: add, del: del})
+			}
 		default:
 			// TODO(ik): Implement String() on those options to have a nice print.
 			// %v is difficult to figure what's what, %+v print private fields and
@@ -2589,3 +2627,24 @@ addLoop:
 
 	return add, remove
 }
+
+func diffProxiesTrustedKeys(old, new []*ProxyConfig) ([]string, []string) {
+	var add []string
+	var del []string
+	// Both "old" and "new" lists should be small...
+	for _, op := range old {
+		if !slices.ContainsFunc(new, func(pc *ProxyConfig) bool {
+			return pc.Key == op.Key
+		}) {
+			del = append(del, op.Key)
+		}
+	}
+	for _, np := range new {
+		if !slices.ContainsFunc(old, func(pc *ProxyConfig) bool {
+			return pc.Key == np.Key
+		}) {
+			add = append(add, np.Key)
+		}
+	}
+	return add, del
+}
diff --git a/server/server.go b/server/server.go
index 99f7dd14..2875425d 100644
--- a/server/server.go
+++ b/server/server.go
@@ -133,6 +133,7 @@ type Info struct {
 	ConnectInfo       bool     `json:"connect_info,omitempty"`   // When true this is the server INFO response to CONNECT
 	RemoteAccount     string   `json:"remote_account,omitempty"` // Lets the client or leafnode side know the remote account that they bind to.
 	IsSystemAccount   bool     `json:"acc_is_sys,omitempty"`     // Indicates if the account is a system account.
+	JSApiLevel        int      `json:"api_lvl,omitempty"`
 
 	// Route Specific
 	Import        *SubjectPermission `json:"import,omitempty"`
@@ -371,6 +372,11 @@ type Server struct {
 	// Controls whether or not the account NRG capability is set in statsz.
 	// Currently used by unit tests to simulate nodes not supporting account NRG.
 	accountNRGAllowed atomic.Bool
+
+	// List of proxies trusted keys in `KeyPair` form so we can do signature
+	// verification when processing incoming proxy connections.
+	proxiesKeyPairs []nkeys.KeyPair
+	proxiedConns    map[string]map[uint64]*client
 }
 
 // For tracking JS nodes.
@@ -732,6 +738,7 @@ func NewServer(opts *Options) (*Server, error) {
 		Headers:      !opts.NoHeaderSupport,
 		Cluster:      opts.Cluster.Name,
 		Domain:       opts.JetStreamDomain,
+		JSApiLevel:   JSApiLevel,
 	}
 
 	if tlsReq && !info.TLSRequired {
@@ -800,6 +807,11 @@ func NewServer(opts *Options) (*Server, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	// If there are proxies trusted public keys in the configuration
+	// this will fill create the corresponding list of nkeys.KeyPair
+	// that we can use for signature verification.
+	s.processProxiesTrustedKeys()
+
 	// Place ourselves in the JetStream nodeInfo if needed.
 	if opts.JetStream {
 		ourNode := getHash(serverName)
@@ -1125,6 +1137,10 @@ func validateOptions(o *Options) error {
 	if err := validateAuth(o); err != nil {
 		return err
 	}
+	// Check that proxies is properly configured.
+	if err := validateProxies(o); err != nil {
+		return err
+	}
 	// Check that gateway is properly configured. Returns no error
 	// if there is no gateway defined.
 	if err := validateGatewayOptions(o); err != nil {
@@ -3589,6 +3605,7 @@ func (s *Server) removeClient(c *client) {
 		if c.kind == CLIENT && c.opts.Protocol >= ClientProtoInfo {
 			updateProtoInfoCount = true
 		}
+		proxyKey := c.proxyKey
 		c.mu.Unlock()
 
 		s.mu.Lock()
@@ -3596,6 +3613,9 @@ func (s *Server) removeClient(c *client) {
 		if updateProtoInfoCount {
 			s.cproto--
 		}
+		if proxyKey != _EMPTY_ {
+			s.removeProxiedConn(proxyKey, cid)
+		}
 		s.mu.Unlock()
 	case ROUTER:
 		s.removeRoute(c)
@@ -3606,6 +3626,18 @@ func (s *Server) removeClient(c *client) {
 	}
 }
 
+// Remove the connection with id `cid` from the map of connections
+// under the public key `key` of the trusted proxies.
+//
+// Server lock must be held on entry.
+func (s *Server) removeProxiedConn(key string, cid uint64) {
+	conns := s.proxiedConns[key]
+	delete(conns, cid)
+	if len(conns) == 0 {
+		delete(s.proxiedConns, key)
+	}
+}
+
 func (s *Server) removeFromTempClients(cid uint64) {
 	s.grMu.Lock()
 	delete(s.grTmpClients, cid)
diff --git a/server/store.go b/server/store.go
index 7e613d8e..da4e5953 100644
--- a/server/store.go
+++ b/server/store.go
@@ -122,7 +122,7 @@ type StreamStore interface {
 	RegisterStorageRemoveMsg(StorageRemoveMsgHandler)
 	RegisterSubjectDeleteMarkerUpdates(SubjectDeleteMarkerUpdateHandler)
 	UpdateConfig(cfg *StreamConfig) error
-	Delete() error
+	Delete(inline bool) error
 	Stop() error
 	ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerStore, error)
 	AddConsumer(o ConsumerStore) error
diff --git a/server/stream.go b/server/stream.go
index ecdcae14..ac71a0f8 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -116,10 +116,6 @@ type StreamConfig struct {
 	// AllowAtomicPublish allows atomic batch publishing into the stream.
 	AllowAtomicPublish bool `json:"allow_atomic"`
 
-	// AllowAsyncFlush allows replicated streams to asynchronously flush
-	// to the stream, improving throughput.
-	AllowAsyncFlush bool `json:"allow_async_flush"`
-
 	// Metadata is additional metadata for the Stream.
 	Metadata map[string]string `json:"metadata,omitempty"`
 }
@@ -798,12 +794,12 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		}
 	}
 	fsCfg.StoreDir = storeDir
-	// Async flushing is only allowed if the stream has a sync log backing it.
-	fsCfg.AsyncFlush = config.AllowAsyncFlush && config.Replicas > 1
 	// Grab configured sync interval.
 	fsCfg.SyncInterval = s.getOpts().SyncInterval
 	fsCfg.SyncAlways = s.getOpts().SyncAlways
 	fsCfg.Compression = config.Compression
+	// Async flushing is only allowed if the stream has a sync log backing it.
+	fsCfg.AsyncFlush = !fsCfg.SyncAlways && config.Replicas > 1
 
 	if err := mset.setupStore(fsCfg); err != nil {
 		mset.stop(true, false)
@@ -812,9 +808,9 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 
 	// Create our pubAck template here. Better than json marshal each time on success.
 	if domain := s.getOpts().JetStreamDomain; domain != _EMPTY_ {
-		mset.pubAck = []byte(fmt.Sprintf("{%q:%q, %q:%q, %q:", "stream", cfg.Name, "domain", domain, "seq"))
+		mset.pubAck = fmt.Appendf(nil, "{%q:%q,%q:%q,%q:", "stream", cfg.Name, "domain", domain, "seq")
 	} else {
-		mset.pubAck = []byte(fmt.Sprintf("{%q:%q, %q:", "stream", cfg.Name, "seq"))
+		mset.pubAck = fmt.Appendf(nil, "{%q:%q,%q:", "stream", cfg.Name, "seq")
 	}
 	end := len(mset.pubAck)
 	mset.pubAck = mset.pubAck[:end:end]
@@ -1063,7 +1059,7 @@ func (mset *stream) setLeader(isLeader bool) error {
 		// Stop responding to sync requests.
 		mset.stopClusterSubs()
 		// Unsubscribe from direct stream.
-		mset.unsubscribeToStream(false)
+		mset.unsubscribeToStream(false, false)
 		// Clear catchup state
 		mset.clearAllCatchupPeers()
 	}
@@ -2268,7 +2264,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool,
 
 	// If atomic publish is disabled, delete any in-progress batches.
 	if !cfg.AllowAtomicPublish {
-		mset.deleteInflightBatches()
+		mset.deleteInflightBatches(false)
 	}
 
 	// Now update config and store's version of our config.
@@ -4204,7 +4200,7 @@ func (mset *stream) removeInternalConsumer(si *sourceInfo) {
 
 // Will unsubscribe from the stream.
 // Lock should be held.
-func (mset *stream) unsubscribeToStream(stopping bool) error {
+func (mset *stream) unsubscribeToStream(stopping, shuttingDown bool) error {
 	for _, subject := range mset.cfg.Subjects {
 		mset.unsubscribeInternal(subject)
 	}
@@ -4217,7 +4213,7 @@ func (mset *stream) unsubscribeToStream(stopping bool) error {
 		mset.stopSourceConsumers()
 	}
 	// Clear batching state.
-	mset.deleteInflightBatches()
+	mset.deleteInflightBatches(shuttingDown)
 
 	// In case we had direct get subscriptions.
 	if mset.directLeaderSub == nil {
@@ -4235,11 +4231,16 @@ func (mset *stream) unsubscribeToStream(stopping bool) error {
 }
 
 // Lock should be held.
-func (mset *stream) deleteInflightBatches() {
+func (mset *stream) deleteInflightBatches(shuttingDown bool) {
 	if mset.batches != nil {
 		mset.batches.mu.Lock()
-		for _, b := range mset.batches.group {
-			b.store.Delete()
+		for batchId, b := range mset.batches.group {
+			// If shutting down, do fixup during startup. In-memory batches don't require manual cleanup.
+			if shuttingDown {
+				b.stopLocked()
+			} else {
+				b.cleanupLocked(batchId, mset.batches)
+			}
 		}
 		mset.batches.mu.Unlock()
 		mset.batches = nil
@@ -4660,7 +4661,12 @@ func (mset *stream) processDirectGetRequest(_ *subscription, c *client, _ *Accou
 	if len(reply) == 0 {
 		return
 	}
-	_, msg := c.msgParts(rmsg)
+	hdr, msg := c.msgParts(rmsg)
+	if errorOnRequiredApiLevel(hdr) {
+		hdr := []byte("NATS/1.0 412 Required Api Level\r\n\r\n")
+		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+		return
+	}
 	if len(msg) == 0 {
 		hdr := []byte("NATS/1.0 408 Empty Request\r\n\r\n")
 		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
@@ -4730,7 +4736,13 @@ func (mset *stream) processDirectGetLastBySubjectRequest(_ *subscription, c *cli
 		return
 	}
 	var minLastSeq uint64
-	_, msg := c.msgParts(rmsg)
+	hdr, msg := c.msgParts(rmsg)
+	if errorOnRequiredApiLevel(hdr) {
+		hdr := []byte("NATS/1.0 412 Required Api Level\r\n\r\n")
+		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+		return
+	}
+
 	// This version expects no payload, unless min last seq is specified.
 	if len(msg) != 0 {
 		var req JSApiMsgGetRequest
@@ -5131,6 +5143,18 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 
 	var resp = &JSPubAckResponse{}
 
+	var batchId string
+	var batchSeq uint64
+	if len(hdr) > 0 {
+		// Populate batch details.
+		if batchId = getBatchId(hdr); batchId != _EMPTY_ {
+			batchSeq, _ = getBatchSequence(hdr)
+			// Disable consistency checking if this was already done
+			// earlier as part of the batch consistency check.
+			canConsistencyCheck = traceOnly
+		}
+	}
+
 	// Bail here if sealed.
 	if canConsistencyCheck && isSealed {
 		mset.mu.Unlock()
@@ -5192,17 +5216,9 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	// Process additional msg headers if still present.
 	var msgId string
 	var incr *big.Int
-	var batchId string
-	var batchSeq uint64
 	var rollupSub, rollupAll bool
 
 	if len(hdr) > 0 {
-
-		// Populate batch details.
-		if batchId = getBatchId(hdr); batchId != _EMPTY_ {
-			batchSeq, _ = getBatchSequence(hdr)
-		}
-
 		// Certain checks have already been performed if in clustered mode, so only check if not.
 		// Note, for cluster mode but with message tracing (without message delivery), we need
 		// to do this check here since it was not done in processClusteredInboundMsg().
@@ -5825,6 +5841,345 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	return nil
 }
 
+// processJetStreamBatchMsg processes a JetStream message that's part of an atomic batch publish.
+// Handles constraints around the batch, storing messages, doing consistency checks, and performing the commit.
+func (mset *stream) processJetStreamBatchMsg(batchId, subject, reply string, hdr, msg []byte, mt *msgTrace) (retErr error) {
+	// For possible error response.
+	var response []byte
+
+	mset.mu.RLock()
+	canRespond := !mset.cfg.NoAck && len(reply) > 0
+	name, stype := mset.cfg.Name, mset.cfg.Storage
+	discard, discardNewPer, maxMsgs, maxMsgsPer, maxBytes := mset.cfg.Discard, mset.cfg.DiscardNewPer, mset.cfg.MaxMsgs, mset.cfg.MaxMsgsPer, mset.cfg.MaxBytes
+	s, js, jsa, st, r, tierName, outq, node := mset.srv, mset.js, mset.jsa, mset.cfg.Storage, mset.cfg.Replicas, mset.tier, mset.outq, mset.node
+	maxMsgSize, lseq := int(mset.cfg.MaxMsgSize), mset.lseq
+	isLeader, isClustered, isSealed, allowRollup, denyPurge, allowTTL, allowMsgCounter, allowAtomicPublish := mset.isLeader(), mset.isClustered(), mset.cfg.Sealed, mset.cfg.AllowRollup, mset.cfg.DenyPurge, mset.cfg.AllowMsgTTL, mset.cfg.AllowMsgCounter, mset.cfg.AllowAtomicPublish
+	mset.mu.RUnlock()
+
+	// If message tracing (with message delivery), we will need to send the
+	// event on exit in case there was an error (if message was not proposed).
+	// Otherwise, the event will be sent from processJetStreamMsg when
+	// invoked by the leader (from applyStreamEntries).
+	if mt != nil {
+		defer func() {
+			if retErr != nil {
+				mt.sendEventFromJetStream(retErr)
+			}
+		}()
+	}
+
+	// Check that we are the leader. This can be false if we have scaled up from an R1 that had inbound queued messages.
+	if !isLeader {
+		return NewJSClusterNotLeaderError()
+	}
+
+	// Bail here if sealed.
+	if isSealed {
+		var resp = JSPubAckResponse{PubAck: &PubAck{Stream: mset.name()}, Error: NewJSStreamSealedError()}
+		b, _ := json.Marshal(resp)
+		mset.outq.sendMsg(reply, b)
+		return NewJSStreamSealedError()
+	}
+
+	// Check here pre-emptively if we have exceeded this server limits.
+	if js.limitsExceeded(stype) {
+		s.resourcesExceededError(stype)
+		if canRespond {
+			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: NewJSInsufficientResourcesError()})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
+		}
+		// Stepdown regardless.
+		if node := mset.raftNode(); node != nil {
+			node.StepDown()
+		}
+		return NewJSInsufficientResourcesError()
+	}
+
+	// Check here pre-emptively if we have exceeded our account limits.
+	if exceeded, err := jsa.wouldExceedLimits(st, tierName, r, subject, hdr, msg); exceeded {
+		if err == nil {
+			err = NewJSAccountResourcesExceededError()
+		}
+		s.RateLimitWarnf("JetStream account limits exceeded for '%s': %s", jsa.acc().GetName(), err.Error())
+		if canRespond {
+			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
+			resp.Error = err
+			response, _ = json.Marshal(resp)
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
+		}
+		return err
+	}
+
+	// Check msgSize if we have a limit set there. Again this works if it goes through but better to be pre-emptive.
+	// Subtract to prevent against overflows.
+	if maxMsgSize >= 0 && (len(hdr) > maxMsgSize || len(msg) > maxMsgSize-len(hdr)) {
+		err := fmt.Errorf("JetStream message size exceeds limits for '%s > %s'", jsa.acc().Name, mset.cfg.Name)
+		s.RateLimitWarnf("%s", err.Error())
+		if canRespond {
+			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
+			resp.Error = NewJSStreamMessageExceedsMaximumError()
+			response, _ = json.Marshal(resp)
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
+		}
+		return err
+	}
+
+	if !allowAtomicPublish {
+		err := NewJSAtomicPublishDisabledError()
+		if canRespond {
+			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
+		}
+		return err
+	}
+
+	// Batch ID is too long.
+	if len(batchId) > 64 {
+		err := NewJSAtomicPublishInvalidBatchIDError()
+		if canRespond {
+			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
+		}
+		return err
+	}
+
+	batchSeq, exists := getBatchSequence(hdr)
+	if !exists {
+		err := NewJSAtomicPublishMissingSeqError()
+		if canRespond {
+			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
+		}
+		return err
+	}
+	commit := len(sliceHeader(JSBatchCommit, hdr)) != 0
+
+	mset.mu.Lock()
+	if mset.batches == nil {
+		mset.batches = &batching{
+			group: make(map[string]*batchGroup, 1),
+		}
+	}
+	batches := mset.batches
+	mset.mu.Unlock()
+
+	respondIncompleteBatch := func() error {
+		err := NewJSAtomicPublishIncompleteBatchError()
+		if canRespond {
+			buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
+		}
+		return err
+	}
+
+	// Get batch.
+	batches.mu.Lock()
+	b, ok := batches.group[batchId]
+	if !ok {
+		if batchSeq != 1 {
+			batches.mu.Unlock()
+			return respondIncompleteBatch()
+		}
+
+		// Limits.
+		maxInflightPerStream := streamMaxBatchInflightPerStream
+		maxInflightTotal := streamMaxBatchInflightTotal
+		opts := s.getOpts()
+		if opts.JetStreamLimits.MaxBatchInflightPerStream > 0 {
+			maxInflightPerStream = opts.JetStreamLimits.MaxBatchInflightPerStream
+		}
+		if opts.JetStreamLimits.MaxBatchInflightTotal > 0 {
+			maxInflightTotal = opts.JetStreamLimits.MaxBatchInflightTotal
+		}
+
+		// Confirm we can facilitate an additional batch.
+		if len(batches.group)+1 > maxInflightPerStream {
+			batches.mu.Unlock()
+			return respondIncompleteBatch()
+		}
+
+		// Confirm we'll not exceed the server limit.
+		if globalInflightBatches.Add(1) > int32(maxInflightTotal) {
+			globalInflightBatches.Add(-1)
+			batches.mu.Unlock()
+			return respondIncompleteBatch()
+		}
+
+		var err error
+		if b, err = batches.newBatchGroup(mset, batchId); err != nil {
+			globalInflightBatches.Add(-1)
+			batches.mu.Unlock()
+			return respondIncompleteBatch()
+		}
+		batches.group[batchId] = b
+	}
+
+	// Detect gaps.
+	b.lseq++
+	if b.lseq != batchSeq {
+		b.cleanupLocked(batchId, batches)
+		batches.mu.Unlock()
+		return respondIncompleteBatch()
+	}
+
+	// Confirm the batch doesn't exceed the allowed size.
+	maxSize := streamMaxBatchSize
+	if maxBatchSize := s.getOpts().JetStreamLimits.MaxBatchSize; maxBatchSize > 0 {
+		maxSize = maxBatchSize
+	}
+	if batchSeq > uint64(maxSize) {
+		b.cleanupLocked(batchId, batches)
+		batches.mu.Unlock()
+		return respondIncompleteBatch()
+	}
+
+	// Persist, but optimize if we're committing because we already know last.
+	// If the underlying store is file-based we need to persist everything to survive hard kills, because we're R1.
+	if !commit || b.store.Type() == FileStorage {
+		seq, _, err := b.store.StoreMsg(subject, hdr, msg, 0)
+		if err != nil || seq != batchSeq {
+			b.cleanupLocked(batchId, batches)
+			batches.mu.Unlock()
+			return respondIncompleteBatch()
+		}
+	}
+	if !commit {
+		batches.mu.Unlock()
+		return nil
+	}
+
+	// Ensure the batch is prepared for the commit and will not be cleaned up while committing.
+	if !b.readyForCommit() {
+		// Don't do cleanup, this is already done.
+		batches.mu.Unlock()
+		return respondIncompleteBatch()
+	}
+
+	// Proceed with proposing the batch.
+	ts := time.Now().UnixNano()
+
+	// Need to hold this lock even if we're not clustered, because we'll
+	// be accessing state that requires this lock (even if it's empty).
+	mset.clMu.Lock()
+	rollback := func(seq uint64) {
+		if isClustered {
+			// TODO(mvv): reset in-memory expected header maps
+			mset.clseq -= seq - 1
+		}
+		mset.clMu.Unlock()
+	}
+
+	errorOnUnsupported := func(seq uint64, header string) *ApiError {
+		apiErr := NewJSAtomicPublishUnsupportedHeaderBatchError(header)
+		rollback(seq)
+		b.cleanupLocked(batchId, batches)
+		batches.mu.Unlock()
+		if canRespond {
+			buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: apiErr})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
+		}
+		return apiErr
+	}
+
+	var (
+		entries []*Entry
+		bsubj   string
+		bhdr    []byte
+		bmsg    []byte
+		apiErr  *ApiError
+		err     error
+		smv     StoreMsg
+		sm      *StoreMsg
+		sz      int
+	)
+
+	diff := &batchStagedDiff{}
+	for seq := uint64(1); seq <= batchSeq; seq++ {
+		if seq == batchSeq && b.store.Type() != FileStorage {
+			bsubj, bhdr, bmsg = subject, hdr, msg
+		} else if sm, err = b.store.LoadMsg(seq, &smv); sm != nil && err == nil {
+			bsubj, bhdr, bmsg = sm.subj, sm.hdr, sm.msg
+		} else {
+			rollback(seq)
+			b.cleanupLocked(batchId, batches)
+			batches.mu.Unlock()
+			return respondIncompleteBatch()
+		}
+
+		// Reject unsupported headers.
+		if msgId := getMsgId(bhdr); msgId != _EMPTY_ {
+			return errorOnUnsupported(seq, JSMsgId)
+		}
+		if getExpectedLastMsgId(hdr) != _EMPTY_ {
+			return errorOnUnsupported(seq, JSExpectedLastMsgId)
+		}
+
+		if bhdr, bmsg, _, apiErr, err = checkMsgHeadersPreClusteredProposal(diff, mset, subject, bhdr, bmsg, false, name, jsa, allowRollup, denyPurge, allowTTL, allowMsgCounter, discard, discardNewPer, maxMsgSize, maxMsgs, maxMsgsPer, maxBytes); err != nil {
+			rollback(seq)
+			b.cleanupLocked(batchId, batches)
+			batches.mu.Unlock()
+			if canRespond {
+				buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: apiErr})
+				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
+			}
+			return err
+		}
+
+		if isClustered {
+			var _reply string
+			if seq == batchSeq {
+				_reply = reply
+			}
+			esm := encodeStreamMsgAllowCompress(bsubj, _reply, bhdr, bmsg, mset.clseq, ts, false)
+			entries = append(entries, newEntry(EntryNormal, esm))
+			mset.clseq++
+			sz += len(esm)
+		}
+	}
+
+	// Commit batch.
+	if !isClustered {
+		for seq := uint64(1); seq <= batchSeq; seq++ {
+			if seq == batchSeq && b.store.Type() != FileStorage {
+				bsubj, bhdr, bmsg = subject, hdr, msg
+			} else if sm, err = b.store.LoadMsg(seq, &smv); sm != nil && err == nil {
+				bsubj, bhdr, bmsg = sm.subj, sm.hdr, sm.msg
+			} else {
+				// Should not happen, we've already checked this message existed while doing consistency checks.
+				// We'll just exit here, the batch is already inconsistent without the message at this sequence.
+				// No use in trying to still store the rest.
+				break
+			}
+			var _reply string
+			if seq == batchSeq {
+				_reply = reply
+			}
+			mset.processJetStreamMsg(bsubj, _reply, bhdr, bmsg, 0, 0, mt, false)
+		}
+	} else {
+		// Do proposal.
+		diff.commit(mset)
+		// TODO(mvv): replace with individual `node.Propose`?
+		if err := node.ProposeMulti(entries); err == nil {
+			mset.trackReplicationTraffic(node, sz, r)
+		} else {
+			// TODO(mvv): reset in-memory expected header maps
+			mset.clseq -= batchSeq
+		}
+
+		// Check to see if we are being overrun.
+		// TODO(dlc) - Make this a limit where we drop messages to protect ourselves, but allow to be configured.
+		if mset.clseq-(lseq+mset.clfs) > streamLagWarnThreshold {
+			lerr := fmt.Errorf("JetStream stream '%s > %s' has high message lag", jsa.acc().Name, name)
+			s.RateLimitWarnf("%s", lerr.Error())
+		}
+	}
+	mset.clMu.Unlock()
+	b.cleanupLocked(batchId, batches)
+	batches.mu.Unlock()
+	return nil
+}
+
 // Used to signal inbound message to registered consumers.
 type cMsg struct {
 	seq  uint64
@@ -6122,7 +6477,9 @@ func (mset *stream) internalLoop() {
 			ims := msgs.pop()
 			for _, im := range ims {
 				// If we are clustered we need to propose this message to the underlying raft group.
-				if isClustered {
+				if batchId := getBatchId(im.hdr); batchId != _EMPTY_ {
+					mset.processJetStreamBatchMsg(batchId, im.subj, im.rply, im.hdr, im.msg, im.mt)
+				} else if isClustered {
 					mset.processClusteredInboundMsg(im.subj, im.rply, im.hdr, im.msg, im.mt, false)
 				} else {
 					mset.processJetStreamMsg(im.subj, im.rply, im.hdr, im.msg, 0, 0, im.mt, false)
@@ -6190,6 +6547,8 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 		return NewJSNotEnabledForAccountError()
 	}
 
+	isShuttingDown := js.isShuttingDown()
+
 	// Remove from our account map first.
 	jsa.mu.Lock()
 	delete(jsa.streams, name)
@@ -6212,7 +6571,7 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	// Stop responding to sync requests.
 	mset.stopClusterSubs()
 	// Unsubscribe from direct stream.
-	mset.unsubscribeToStream(true)
+	mset.unsubscribeToStream(true, isShuttingDown)
 
 	// Our info sub if we spun it up.
 	if mset.infoSub != nil {
@@ -6243,7 +6602,6 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	}
 	mset.mu.Unlock()
 
-	isShuttingDown := js.isShuttingDown()
 	for _, o := range obs {
 		if !o.isClosed() {
 			// Third flag says do not broadcast a signal.
@@ -6332,7 +6690,7 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	if deleteFlag {
 		if store != nil {
 			// Ignore errors.
-			store.Delete()
+			store.Delete(false)
 		}
 		// Release any resources.
 		js.releaseStreamResources(&mset.cfg)
diff --git a/server/sysmem/mem_solaris.go b/server/sysmem/mem_solaris.go
new file mode 100644
index 00000000..afba0364
--- /dev/null
+++ b/server/sysmem/mem_solaris.go
@@ -0,0 +1,37 @@
+// Copyright 2025 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build illumos || solaris
+
+package sysmem
+
+import (
+	"golang.org/x/sys/unix"
+)
+
+const (
+	_SC_PHYS_PAGES = 500
+	_SC_PAGESIZE   = 11
+)
+
+func Memory() int64 {
+	pages, err := unix.Sysconf(_SC_PHYS_PAGES)
+	if err != nil {
+		return 0
+	}
+	pageSize, err := unix.Sysconf(_SC_PAGESIZE)
+	if err != nil {
+		return 0
+	}
+	return int64(pages) * int64(pageSize)
+}



Test patch:
diff --git a/server/auth_callout_test.go b/server/auth_callout_test.go
index 7021d7c5..c7586d30 100644
--- a/server/auth_callout_test.go
+++ b/server/auth_callout_test.go
@@ -241,6 +241,11 @@ func TestAuthCalloutBasics(t *testing.T) {
 			}
 			ujwt := createAuthUser(t, user, _EMPTY_, globalAccountName, "", nil, 10*time.Minute, &j)
 			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
+		} else if opts.Username == "proxy" {
+			var j jwt.UserPermissionLimits
+			j.ProxyRequired = true
+			ujwt := createAuthUser(t, user, _EMPTY_, globalAccountName, "", nil, 10*time.Minute, &j)
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
 		} else {
 			// Nil response signals no authentication.
 			m.Respond(nil)
@@ -252,6 +257,9 @@ func TestAuthCalloutBasics(t *testing.T) {
 	// This one should fail since bad password.
 	at.RequireConnectError(nats.UserInfo("dlc", "xxx"))
 
+	// This one should fail because it will require to be proxied and it is not.
+	at.RequireConnectError(nats.UserInfo("proxy", "xxx"))
+
 	// This one will use callout since not defined in server config.
 	nc := at.Connect(nats.UserInfo("dlc", "zzz"))
 	defer nc.Close()
@@ -1355,8 +1363,13 @@ func TestAuthCalloutAuthErrEvents(t *testing.T) {
 			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
 		} else if opts.Username == "dlc" {
 			m.Respond(serviceResponse(t, user, si.ID, "", "WRONG PASSWORD", 0))
-		} else {
+		} else if opts.Username == "rip" {
 			m.Respond(serviceResponse(t, user, si.ID, "", "BAD CREDS", 0))
+		} else if opts.Username == "proxy" {
+			var j jwt.UserPermissionLimits
+			j.ProxyRequired = true
+			ujwt := createAuthUser(t, user, _EMPTY_, "FOO", "", nil, 0, &j)
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
 		}
 	}
 
@@ -1382,13 +1395,19 @@ func TestAuthCalloutAuthErrEvents(t *testing.T) {
 		err = json.Unmarshal(m.Data, &dm)
 		require_NoError(t, err)
 
-		if !strings.Contains(dm.Reason, reason) {
-			t.Fatalf("Expected %q reason, but got %q", reason, dm.Reason)
+		// Convert both reasons to lower case to do the comparison.
+		dmr := strings.ToLower(dm.Reason)
+		r := strings.ToLower(reason)
+
+		if !strings.Contains(dmr, r) {
+			t.Fatalf("Expected %q reason, but got %q", r, dmr)
 		}
 	}
 
 	checkAuthErrEvent("dlc", "xxx", "WRONG PASSWORD")
 	checkAuthErrEvent("rip", "abc", "BAD CREDS")
+	// The auth callout uses as the reason the error string, not a closed state.
+	checkAuthErrEvent("proxy", "proxy", ErrAuthProxyRequired.Error())
 }
 
 func TestAuthCalloutConnectEvents(t *testing.T) {
@@ -2402,3 +2421,74 @@ func TestAuthCalloutLeafNodeAndConfigMode(t *testing.T) {
 	}
 
 }
+
+func TestAuthCalloutProxyRequiredInUserNotInAuthJWT(t *testing.T) {
+	conf := `
+		listen: "127.0.0.1:-1"
+		server_name: A
+		accounts {
+			AUTH: {
+				users: [ { user: auth, password: auth } ]
+			}
+			APP: {
+				users: [
+					{ user: user, password: pwd }
+					{ user: proxy, password: pwd, proxy_required: true }
+				]
+			}
+			SYS: {}
+		}
+		system_account: SYS
+
+		authorization {
+			timeout: 1s
+			auth_callout {
+				issuer: "ABJHLOVMPA4CI6R5KLNGOB4GSLNIY7IOUPAJC4YFNDLQVIOBYQGUWVLA"
+				auth_users: [ auth ]
+				account: AUTH
+			}
+		}
+	`
+	var invoked atomic.Int32
+	handler := func(m *nats.Msg) {
+		invoked.Add(1)
+		user, si, ci, opts, _ := decodeAuthRequest(t, m.Data)
+		require_True(t, si.Name == "A")
+		require_True(t, ci.Host == "127.0.0.1")
+		if opts.Username == "proxy" {
+			// If we don't set a ProxyRequired property explicitly here, but the
+			// user has it, so it should still be rejected.
+			ujwt := createAuthUser(t, user, _EMPTY_, "APP", _EMPTY_, nil, 10*time.Minute, nil)
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, _EMPTY_, 0))
+		} else {
+			m.Respond(nil)
+		}
+	}
+	at := NewAuthTest(t, conf, handler, nats.UserInfo("auth", "auth"))
+	defer at.Cleanup()
+
+	sub, err := at.authClient.SubscribeSync(authErrorAccountEventSubj)
+	require_NoError(t, err)
+
+	// It should fail, even if the auth callout does not require proxy in its JWT.
+	// In other words, we reject if not proxied and the user config or the auth JWT
+	// requires proxy connection.
+	at.RequireConnectError(nats.UserInfo("proxy", "pwd"))
+
+	m, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+
+	var dm DisconnectEventMsg
+	err = json.Unmarshal(m.Data, &dm)
+	require_NoError(t, err)
+
+	// Convert both reasons to lower case to do the comparison.
+	dmr := strings.ToLower(dm.Reason)
+	r := strings.ToLower(ErrAuthProxyRequired.Error())
+	if !strings.Contains(dmr, r) {
+		t.Fatalf("Expected %q reason, but got %q", r, dmr)
+	}
+
+	// Auth callout should have been invoked once.
+	require_Equal(t, 1, int(invoked.Load()))
+}
diff --git a/server/auth_test.go b/server/auth_test.go
index 05c3402f..d0e2ff9a 100644
--- a/server/auth_test.go
+++ b/server/auth_test.go
@@ -16,6 +16,7 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"net/url"
@@ -27,6 +28,7 @@ import (
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 )
 
 func TestUserCloneNilPermissions(t *testing.T) {
@@ -388,3 +390,373 @@ func TestNoAuthUserNoConnectProto(t *testing.T) {
 	time.Sleep(1200 * time.Millisecond)
 	checkClientsCount(t, s, 0)
 }
+
+type captureProxyRequiredLogger struct {
+	DummyLogger
+	ch chan string
+}
+
+func (l *captureProxyRequiredLogger) Debugf(format string, args ...any) {
+	msg := fmt.Sprintf(format, args...)
+	if strings.Contains(msg, ErrAuthProxyRequired.Error()) {
+		select {
+		case l.ch <- msg:
+		default:
+		}
+	}
+}
+
+func TestAuthProxyRequired(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		authorization {
+			user: user
+			password: pwd
+			proxy_required: true
+		}
+	`))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	l := &captureProxyRequiredLogger{ch: make(chan string, 1)}
+	s.SetLogger(l, true, false)
+
+	_, err := nats.Connect(s.ClientURL(), nats.UserInfo("user", "pwd"))
+	require_True(t, errors.Is(err, nats.ErrAuthorization))
+
+	checkLog := func() {
+		t.Helper()
+		select {
+		case <-l.ch:
+			return
+		case <-time.After(time.Second):
+			t.Fatal("Did not get log statement")
+		}
+	}
+	checkLog()
+
+	drainLog := func() {
+		for {
+			select {
+			case <-l.ch:
+			default:
+				return
+			}
+		}
+	}
+	s.Shutdown()
+	drainLog()
+
+	nkUsr1, err := nkeys.CreateUser()
+	require_NoError(t, err)
+	nkPub1, err := nkUsr1.PublicKey()
+	require_NoError(t, err)
+
+	nkUsr2, err := nkeys.CreateUser()
+	require_NoError(t, err)
+	nkPub2, err := nkUsr2.PublicKey()
+	require_NoError(t, err)
+
+	conf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		authorization {
+			users: [
+				{user: user1, password: pwd1}
+				{user: user2, password: pwd2, proxy_required: true}
+				{user: user3, password: pwd3, proxy_required: false}
+				{nkey: "%s", proxy_required: true}
+				{nkey: "%s", proxy_required: false}
+			]
+		}
+	`, nkPub1, nkPub2))
+	s, _ = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	checkClients := func() {
+		t.Helper()
+		// Should connect ok.
+		nc := natsConnect(t, s.ClientURL(), nats.UserInfo("user1", "pwd1"))
+		nc.Close()
+
+		// Should not, since it requires going through proxy.
+		_, err = nats.Connect(s.ClientURL(), nats.UserInfo("user2", "pwd2"))
+		require_True(t, errors.Is(err, nats.ErrAuthorization))
+		checkLog()
+
+		// Should connect ok.
+		nc = natsConnect(t, s.ClientURL(), nats.UserInfo("user3", "pwd3"))
+		nc.Close()
+
+		// Should not, since it requires going through proxy.
+		_, err = nats.Connect(s.ClientURL(), nats.Nkey(nkPub1, func(nonce []byte) ([]byte, error) {
+			return nkUsr1.Sign(nonce)
+		}))
+		require_True(t, errors.Is(err, nats.ErrAuthorization))
+		checkLog()
+
+		// Should connect ok.
+		nc = natsConnect(t, s.ClientURL(), nats.Nkey(nkPub2, func(nonce []byte) ([]byte, error) {
+			return nkUsr2.Sign(nonce)
+		}))
+		nc.Close()
+	}
+	checkClients()
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		accounts {
+			A {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+					{nkey: "%s", proxy_required: true}
+					{nkey: "%s", proxy_required: false}
+				]
+			}
+			SYS { users: [ {user:sys, password: pwd} ] }
+		}
+		system_account: SYS
+	`, nkPub1, nkPub2))
+	s, _ = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	snc := natsConnect(t, s.ClientURL(), nats.UserInfo("sys", "pwd"))
+	defer snc.Close()
+	sub := natsSubSync(t, snc, "$SYS.SERVER.*.CLIENT.AUTH.ERR")
+	sub2 := natsSubSync(t, snc, "$SYS.ACCOUNT.A.DISCONNECT")
+	natsFlush(t, snc)
+
+	checkClients()
+
+	// We should get 2 authentication error messages, saying that
+	// the reason is "ProxyRequired".
+	for range 2 {
+		var de DisconnectEventMsg
+		msg := natsNexMsg(t, sub, time.Second)
+		err = json.Unmarshal(msg.Data, &de)
+		require_NoError(t, err)
+		require_Equal(t, de.Reason, ProxyRequired.String())
+	}
+
+	// We should get 3 disconnect events since only 3 have connected
+	// ok and then just closed.
+	for range 3 {
+		var de DisconnectEventMsg
+		msg := natsNexMsg(t, sub2, time.Second)
+		err = json.Unmarshal(msg.Data, &de)
+		require_NoError(t, err)
+		require_Equal(t, de.Reason, ClientClosed.String())
+	}
+	// Make sure there is no more.
+	if msg, err := sub2.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Should have received only 3 disconnect messages, got another: %s", msg.Data)
+	}
+
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes: {
+			port: -1
+			authorization {
+				user: user
+				password: pwd
+				proxy_required: true
+			}
+		}
+	`))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	lconf := createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [ {url: "nats://user:pwd@127.0.0.1:%d"} ]
+		}
+	`, o.LeafNode.Port))
+	leaf, _ := RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes: {
+			port: -1
+			reconnect_interval: "50ms"
+			authorization {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+				]
+			}
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	lconf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [ {url: "nats://user2:pwd@127.0.0.1:%d"} ]
+		}
+	`, o.LeafNode.Port))
+	leaf, _ = RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+	s.Shutdown()
+	drainLog()
+
+	conf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes: {
+			port: -1
+			authorization {
+				nkey: %s
+				proxy_required: true
+			}
+		}
+	`, nkPub1))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	nkSeed1, err := nkUsr1.Seed()
+	require_NoError(t, err)
+
+	lconf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [
+				{
+					url: "nats://127.0.0.1:%d"
+					nkey: "%s"
+				}
+			]
+		}
+	`, o.LeafNode.Port, nkSeed1))
+	leaf, _ = RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+	s.Shutdown()
+	drainLog()
+
+	// Check with operator mode.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		server_name: OP
+		operator = "../test/configs/nkeys/op.jwt"
+		resolver = MEMORY
+		listen: "127.0.0.1:-1"
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	s.SetLogger(l, true, false)
+
+	_, akp := createAccount(s)
+	kp, err := nkeys.CreateUser()
+	require_NoError(t, err)
+	pub, err := kp.PublicKey()
+	require_NoError(t, err)
+	nuc := jwt.NewUserClaims(pub)
+	nuc.ProxyRequired = true
+	ujwt, err := nuc.Encode(akp)
+	require_NoError(t, err)
+
+	lopts := &DefaultTestOptions
+	u, err := url.Parse(fmt.Sprintf("nats://%s:%d", o.LeafNode.Host, o.LeafNode.Port))
+	require_NoError(t, err)
+	remote := &RemoteLeafOpts{URLs: []*url.URL{u}}
+	remote.SignatureCB = func(nonce []byte) (string, []byte, error) {
+		sig, err := kp.Sign(nonce)
+		return ujwt, sig, err
+	}
+	lopts.LeafNode.Remotes = []*RemoteLeafOpts{remote}
+	lopts.LeafNode.ReconnectInterval = 100 * time.Millisecond
+	leaf = RunServer(lopts)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+
+	// Try with an user.
+	drainLog()
+
+	_, err = nats.Connect(s.ClientURL(), createUserCredsEx(t, nuc, akp))
+	require_True(t, errors.Is(err, nats.ErrAuthorization))
+	checkLog()
+
+	drainLog()
+
+	// Try with creds file.
+	kp, err = nkeys.CreateUser()
+	require_NoError(t, err)
+	useed, err := kp.Seed()
+	require_NoError(t, err)
+	pub, err = kp.PublicKey()
+	require_NoError(t, err)
+	nuc = jwt.NewUserClaims(pub)
+	nuc.ProxyRequired = true
+	ujwt, err = nuc.Encode(akp)
+	require_NoError(t, err)
+	credsFile := genCredsFile(t, ujwt, useed)
+
+	lconf = createConfFile(t, fmt.Appendf(nil, `
+		port: -1
+		leafnodes {
+			reconnect_interval: "50ms"
+			remotes [
+				{
+					url: "nats://127.0.0.1:%d"
+					credentials: '%s'
+				}
+			]
+		}
+	`, o.LeafNode.Port, credsFile))
+	leaf, _ = RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	time.Sleep(125 * time.Millisecond)
+	checkLeafNodeConnectedCount(t, leaf, 0)
+	checkLog()
+	leaf.Shutdown()
+
+	s.Shutdown()
+	drainLog()
+}
diff --git a/server/config_check_test.go b/server/config_check_test.go
index 0db9eecb..0c437347 100644
--- a/server/config_check_test.go
+++ b/server/config_check_test.go
@@ -2153,6 +2153,251 @@ func TestConfigCheck(t *testing.T) {
 			errorLine: 3,
 			errorPos:  5,
 		},
+		{
+			name: "Proxies wrong type",
+			config: `
+				port: -1
+				proxies: 123
+			`,
+			err:       errors.New("expected proxies to be a map/struct, got int64"),
+			errorLine: 3,
+			errorPos:  5,
+		},
+		{
+			name: "Proxies unknown field",
+			config: `
+				port: -1
+				proxies: {
+					field1: 123
+				}
+			`,
+			err:       errors.New("unknown field"),
+			errorLine: 4,
+			errorPos:  6,
+		},
+		{
+			name: "Proxies trusted wrong type",
+			config: `
+				port: -1
+				proxies: {
+					trusted: 123
+				}
+			`,
+			err:       errors.New("expected proxies' trusted field to be an array, got int64"),
+			errorLine: 4,
+			errorPos:  6,
+		},
+		{
+			name: "Proxies trusted key wrong type",
+			config: `
+				port: -1
+				proxies: {
+					trusted: [
+						"abc",
+						"def"
+					]
+				}
+			`,
+			err:       errors.New("expected proxies' trusted entry to be a map/struct, got string"),
+			errorLine: 5,
+			errorPos:  8,
+		},
+		{
+			name: "Proxies trusted unknown field",
+			config: `
+				port: -1
+				proxies: {
+					trusted: [
+						{
+							key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+							field1: 123
+						}
+					]
+				}
+			`,
+			err:       errors.New("unknown field"),
+			errorLine: 7,
+			errorPos:  8,
+		},
+		{
+			name: "Proxies trusted key invalid",
+			config: `
+				port: -1
+				proxies: {
+					trusted: [
+						{key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"}
+						{key: "bad1"}
+						{key: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"}
+						{key: "bad2"}
+					]
+				}
+			`,
+			err:       errors.New("invalid proxy key"),
+			errorLine: 6,
+			errorPos:  8,
+		},
+		{
+			name: "Auth user require proxied wrong type",
+			config: `
+				port: -1
+				authorization {
+					user: user
+					password: pwd
+					proxy_required: 123
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 6,
+			errorPos:  6,
+		},
+		{
+			name: "Auth users require proxied wrong type",
+			config: `
+				port: -1
+				authorization {
+					users: [
+						{
+							user: user
+							password: pwd
+							proxy_required: 123
+						}
+					]
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 8,
+			errorPos:  8,
+		},
+		{
+			name: "Auth nkey users require proxied wrong type",
+			config: `
+				port: -1
+				authorization {
+					users: [
+						{
+							nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+							proxy_required: 123
+						}
+					]
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 7,
+			errorPos:  8,
+		},
+		{
+			name: "Account users require proxied wrong type",
+			config: `
+				port: -1
+				accounts {
+					A: {
+						users: [
+							{
+								user: user
+								password: pwd
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 9,
+			errorPos:  9,
+		},
+		{
+			name: "Account nkey users require proxied wrong type",
+			config: `
+				port: -1
+				accounts {
+					A: {
+						users: [
+							{
+								nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 8,
+			errorPos:  9,
+		},
+		{
+			name: "Leafnode user require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						user: user
+						password: pwd
+						proxy_required: 123
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 8,
+			errorPos:  7,
+		},
+		{
+			name: "Leafnode neky user require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+						proxy_required: 123
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 7,
+			errorPos:  7,
+		},
+		{
+			name: "Leafnode users require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						users: [
+							{
+								user: user
+								password: pwd
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 10,
+			errorPos:  9,
+		},
+		{
+			name: "Leafnode nkey users require proxied wrong type",
+			config: `
+				port: -1
+				leafnodes {
+					port: -1
+					authorization {
+						users: [
+							{
+								nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+								proxy_required: 123
+							}
+						]
+					}
+				}
+			`,
+			err:       errors.New("interface conversion"),
+			errorLine: 9,
+			errorPos:  9,
+		},
 	}
 
 	checkConfig := func(config string) error {
diff --git a/server/filestore_test.go b/server/filestore_test.go
index b5e2f33b..91ea25ee 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -2783,7 +2783,7 @@ func TestFileStoreStreamDeleteDirNotEmpty(t *testing.T) {
 		}()
 
 		<-ready
-		if err := fs.Delete(); err != nil {
+		if err := fs.Delete(true); err != nil {
 			t.Fatalf("Delete returned an error: %v", err)
 		}
 	})
@@ -10064,6 +10064,63 @@ func TestFileStoreCompressionAfterTruncate(t *testing.T) {
 	}
 }
 
+func TestFileStoreTruncateRemovedBlock(t *testing.T) {
+	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
+		cfg := StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage}
+		created := time.Now()
+		fs, err := newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+		require_NoError(t, err)
+		defer fs.Stop()
+
+		for i := range 3 {
+			if i > 0 {
+				_, err = fs.newMsgBlockForWrite()
+				require_NoError(t, err)
+			}
+			_, _, err = fs.StoreMsg("foo", nil, nil, 0)
+			require_NoError(t, err)
+		}
+
+		fs.mu.RLock()
+		blks := len(fs.blks)
+		fs.mu.RUnlock()
+		require_Len(t, blks, 3)
+
+		state := fs.State()
+		require_Equal(t, state.Msgs, 3)
+		require_Equal(t, state.FirstSeq, 1)
+		require_Equal(t, state.LastSeq, 3)
+		require_Equal(t, state.NumDeleted, 0)
+
+		removed, err := fs.RemoveMsg(2)
+		require_NoError(t, err)
+		require_True(t, removed)
+
+		fs.mu.RLock()
+		blks = len(fs.blks)
+		fs.mu.RUnlock()
+		require_Len(t, blks, 2)
+
+		fs.mu.RLock()
+		blks = len(fs.blks)
+		fs.mu.RUnlock()
+		require_Len(t, blks, 2)
+
+		state = fs.State()
+		require_Equal(t, state.Msgs, 2)
+		require_Equal(t, state.FirstSeq, 1)
+		require_Equal(t, state.LastSeq, 3)
+		require_Equal(t, state.NumDeleted, 1)
+
+		require_NoError(t, fs.Truncate(2))
+		state = fs.State()
+		require_Equal(t, state.Msgs, 1)
+		require_Equal(t, state.FirstSeq, 1)
+		require_Equal(t, state.LastSeq, 2)
+		require_Equal(t, state.NumDeleted, 1)
+	})
+}
+
 func TestFileStoreAtomicEraseMsg(t *testing.T) {
 	for _, lmb := range []bool{true, false} {
 		t.Run(fmt.Sprintf("lmb=%v", lmb), func(t *testing.T) {
diff --git a/server/jetstream_batching_test.go b/server/jetstream_batching_test.go
index 92c9c39f..65614c44 100644
--- a/server/jetstream_batching_test.go
+++ b/server/jetstream_batching_test.go
@@ -148,10 +148,11 @@ func TestJetStreamAtomicBatchPublish(t *testing.T) {
 
 	for _, storage := range []StorageType{FileStorage, MemoryStorage} {
 		for _, retention := range []RetentionPolicy{LimitsPolicy, InterestPolicy, WorkQueuePolicy} {
-			replicas := 3
-			t.Run(fmt.Sprintf("%s/%s/R%d", storage, retention, replicas), func(t *testing.T) {
-				test(t, storage, retention, replicas)
-			})
+			for _, replicas := range []int{1, 3} {
+				t.Run(fmt.Sprintf("%s/%s/R%d", storage, retention, replicas), func(t *testing.T) {
+					test(t, storage, retention, replicas)
+				})
+			}
 		}
 	}
 }
@@ -306,6 +307,7 @@ func TestJetStreamAtomicBatchPublishLimits(t *testing.T) {
 		}
 	}
 
+	t.Run("R1", func(t *testing.T) { test(t, 1) })
 	t.Run("R3", func(t *testing.T) { test(t, 3) })
 }
 
@@ -352,16 +354,17 @@ func TestJetStreamAtomicBatchPublishDedupeNotAllowed(t *testing.T) {
 
 	for _, storage := range []StorageType{FileStorage, MemoryStorage} {
 		for _, retention := range []RetentionPolicy{LimitsPolicy, InterestPolicy, WorkQueuePolicy} {
-			replicas := 3
-			t.Run(fmt.Sprintf("%s/%s/R%d", storage, retention, replicas), func(t *testing.T) {
-				c := createJetStreamClusterExplicit(t, "R3S", 3)
-				defer c.shutdown()
+			for _, replicas := range []int{1, 3} {
+				t.Run(fmt.Sprintf("%s/%s/R%d", storage, retention, replicas), func(t *testing.T) {
+					c := createJetStreamClusterExplicit(t, "R3S", 3)
+					defer c.shutdown()
 
-				nc, js := jsClientConnect(t, c.randomServer())
-				defer nc.Close()
+					nc, js := jsClientConnect(t, c.randomServer())
+					defer nc.Close()
 
-				test(t, nc, js, storage, retention, replicas)
-			})
+					test(t, nc, js, storage, retention, replicas)
+				})
+			}
 		}
 	}
 }
@@ -641,6 +644,7 @@ func TestJetStreamAtomicBatchPublishDenyHeaders(t *testing.T) {
 		}
 	}
 
+	t.Run("R1", func(t *testing.T) { test(t, 1) })
 	t.Run("R3", func(t *testing.T) { test(t, 3) })
 }
 
@@ -1197,3 +1201,74 @@ func TestJetStreamAtomicBatchPublishExpectedPerSubject(t *testing.T) {
 	t.Run("redundant", func(t *testing.T) { test(t, Redundant) })
 	t.Run("not-first", func(t *testing.T) { test(t, NotFirst) })
 }
+
+func TestJetStreamAtomicBatchPublishSingleServerRecovery(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	cfg := &StreamConfig{
+		Name:               "TEST",
+		Subjects:           []string{"foo"},
+		Storage:            FileStorage,
+		Retention:          LimitsPolicy,
+		Replicas:           1,
+		AllowAtomicPublish: true,
+	}
+	_, err := jsStreamCreate(t, nc, cfg)
+	require_NoError(t, err)
+
+	// Manually construct and store a batch, so it doesn't immediately commit.
+	mset, err := s.globalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	mset.mu.Lock()
+	batches := &batching{}
+	mset.batches = batches
+	mset.mu.Unlock()
+	batches.mu.Lock()
+	b, err := batches.newBatchGroup(mset, "uuid")
+	if err != nil {
+		batches.mu.Unlock()
+		require_NoError(t, err)
+	}
+	hdr1 := genHeader(nil, "Nats-Batch-Id", "uuid")
+	hdr1 = genHeader(hdr1, "Nats-Batch-Sequence", "1")
+	_, _, err = b.store.StoreMsg("foo", hdr1, nil, 0)
+	if err != nil {
+		batches.mu.Unlock()
+		require_NoError(t, err)
+	}
+
+	hdr2 := genHeader(nil, "Nats-Batch-Id", "uuid")
+	hdr2 = genHeader(hdr2, "Nats-Batch-Sequence", "2")
+	hdr2 = genHeader(hdr2, "Nats-Batch-Commit", "1")
+	_, _, err = b.store.StoreMsg("foo", hdr2, nil, 0)
+	if err != nil {
+		batches.mu.Unlock()
+		require_NoError(t, err)
+	}
+	commitReady := b.readyForCommit()
+	batches.mu.Unlock()
+	require_True(t, commitReady)
+
+	// Simulate the first message of the batch is committed.
+	err = mset.processJetStreamMsg("foo", _EMPTY_, hdr1, nil, 0, 0, nil, false)
+	require_NoError(t, err)
+
+	// Simulate a hard kill, upon recovery the rest of the batch should be applied.
+	port := s.opts.Port
+	sd := s.StoreDir()
+	nc.Close()
+	s.Shutdown()
+	s = RunJetStreamServerOnPort(port, sd)
+	defer s.Shutdown()
+
+	mset, err = s.globalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	state := mset.state()
+	require_Equal(t, state.Msgs, 2)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 2)
+}
diff --git a/server/jetstream_benchmark_test.go b/server/jetstream_benchmark_test.go
index 25ab223a..cd75089a 100644
--- a/server/jetstream_benchmark_test.go
+++ b/server/jetstream_benchmark_test.go
@@ -858,14 +858,13 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 		replicas    int
 		messageSize int
 		numSubjects int
-		asyncFlush  bool
 	}{
-		{1, 1, 10, 1, false},   // Single node, 10B messages
-		{1, 1, 1024, 1, false}, // Single node, 1KB messages
-		{3, 3, 10, 1, false},   // 3-nodes cluster, R=3, 10B messages
-		{3, 3, 1024, 1, false}, // 3-nodes cluster, R=3, 1KB messages
-		{3, 3, 10, 1, true},    // 3-nodes cluster, R=3, 10B messages (async flush)
-		{3, 3, 1024, 1, true},  // 3-nodes cluster, R=3, 1KB messages (async flush)
+		{1, 1, 10, 1},   // Single node, 10B messages
+		{1, 1, 1024, 1}, // Single node, 1KB messages
+		{3, 3, 10, 1},   // 3-nodes cluster, R=3, 10B messages
+		{3, 3, 1024, 1}, // 3-nodes cluster, R=3, 1KB messages
+		{3, 3, 10, 1},   // 3-nodes cluster, R=3, 10B messages (async flush)
+		{3, 3, 1024, 1}, // 3-nodes cluster, R=3, 1KB messages (async flush)
 	}
 
 	// All the cases above are run with each of the publisher cases below
@@ -887,10 +886,6 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 			bc.messageSize,
 			bc.numSubjects,
 		)
-		if bc.asyncFlush {
-			name += ",AsyncFlush"
-		}
-
 		b.Run(
 			name,
 			func(b *testing.B) {
@@ -939,11 +934,10 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 								b.Logf("Creating stream with R=%d and %d input subjects", bc.replicas, bc.numSubjects)
 							}
 							_, err = jsStreamCreate(b, nc, &StreamConfig{
-								Name:            streamName,
-								Subjects:        subjects,
-								Replicas:        bc.replicas,
-								Storage:         FileStorage,
-								AllowAsyncFlush: bc.asyncFlush,
+								Name:     streamName,
+								Subjects: subjects,
+								Replicas: bc.replicas,
+								Storage:  FileStorage,
 							})
 							if err != nil {
 								b.Fatalf("Error creating stream: %v", err)
@@ -1942,11 +1936,10 @@ func BenchmarkJetStreamPublishConcurrent(b *testing.B) {
 	replicasCases := []struct {
 		clusterSize int
 		replicas    int
-		asyncFlush  bool
 	}{
-		{1, 1, false},
-		{3, 3, false},
-		{3, 3, true},
+		{1, 1},
+		{3, 3},
+		{3, 3},
 	}
 
 	workload := func(b *testing.B, numPubs int, messageSize int64, clientUrl string) {
@@ -2056,9 +2049,6 @@ func BenchmarkJetStreamPublishConcurrent(b *testing.B) {
 	// benchmark case matrix
 	for _, replicasCase := range replicasCases {
 		title := fmt.Sprintf("N=%d,R=%d", replicasCase.clusterSize, replicasCase.replicas)
-		if replicasCase.asyncFlush {
-			title += ",AsyncFlush"
-		}
 		b.Run(
 			title,
 			func(b *testing.B) {
@@ -2079,11 +2069,10 @@ func BenchmarkJetStreamPublishConcurrent(b *testing.B) {
 
 										// create stream
 										_, err := jsStreamCreate(b, nc, &StreamConfig{
-											Name:            streamName,
-											Subjects:        []string{subject},
-											Replicas:        replicasCase.replicas,
-											Storage:         FileStorage,
-											AllowAsyncFlush: replicasCase.asyncFlush,
+											Name:     streamName,
+											Subjects: []string{subject},
+											Replicas: replicasCase.replicas,
+											Storage:  FileStorage,
 										})
 										if err != nil {
 											b.Fatal(err)
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 515c60d6..961dc493 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -9026,6 +9026,25 @@ func TestJetStreamClusterCreateR3StreamWithOfflineNodes(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterCreateEphemeralConsumerWithOfflineNodes(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	ml := c.leader()
+	nc, js := jsClientConnect(t, ml)
+	defer nc.Close()
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Replicas: 3})
+	require_NoError(t, err)
+
+	// Shutdown a random server.
+	c.randomNonLeader().Shutdown()
+
+	for range 10 {
+		_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{})
+		require_NoError(t, err)
+	}
+}
+
 func TestJetStreamClusterSetPreferredToOnlineNode(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -9059,12 +9078,18 @@ func TestJetStreamClusterSetPreferredToOnlineNode(t *testing.T) {
 }
 
 func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
-	test := func(t *testing.T, replicas int) {
-		supportsAsync := replicas > 1
+	test := func(t *testing.T, syncAlways bool) {
+		supportsAsync := !syncAlways
 
 		c := createJetStreamClusterExplicit(t, "R3S", 3)
 		defer c.shutdown()
 
+		for _, s := range c.servers {
+			s.optsMu.Lock()
+			s.opts.SyncAlways = syncAlways
+			s.optsMu.Unlock()
+		}
+
 		nc, js := jsClientConnect(t, c.randomServer())
 		defer nc.Close()
 
@@ -9101,11 +9126,10 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 		}
 
 		cfg := &StreamConfig{
-			Name:            "TEST",
-			Subjects:        []string{"foo"},
-			Storage:         FileStorage,
-			Replicas:        replicas,
-			AllowAsyncFlush: false,
+			Name:     "TEST",
+			Subjects: []string{"foo"},
+			Storage:  FileStorage,
+			Replicas: 1,
 		}
 		// Test disabled async flush on create.
 		_, err := jsStreamCreate(t, nc, cfg)
@@ -9116,15 +9140,18 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 		checkStoreIsAsync(false)
 
 		// Enabling async flush.
-		cfg.AllowAsyncFlush = true
+		cfg.Replicas = 3
 		_, err = jsStreamUpdate(t, nc, cfg)
 		require_NoError(t, err)
+		checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+			return checkState(t, c, globalAccountName, "TEST")
+		})
 		_, err = js.Publish("foo", nil)
 		require_NoError(t, err)
 		checkStoreIsAsync(supportsAsync)
 
 		// Disabling async flush.
-		cfg.AllowAsyncFlush = false
+		cfg.Replicas = 1
 		_, err = jsStreamUpdate(t, nc, cfg)
 		require_NoError(t, err)
 		_, err = js.Publish("foo", nil)
@@ -9133,7 +9160,7 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 
 		// Test async flush on create.
 		require_NoError(t, js.DeleteStream("TEST"))
-		cfg.AllowAsyncFlush = true
+		cfg.Replicas = 3
 		_, err = jsStreamCreate(t, nc, cfg)
 		require_NoError(t, err)
 		s = c.streamLeader(globalAccountName, "TEST")
@@ -9142,8 +9169,8 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 		checkStoreIsAsync(supportsAsync)
 	}
 
-	t.Run("R1", func(t *testing.T) { test(t, 1) })
-	t.Run("R3", func(t *testing.T) { test(t, 3) })
+	t.Run("Default", func(t *testing.T) { test(t, false) })
+	t.Run("SyncAlways", func(t *testing.T) { test(t, true) })
 }
 
 func TestJetStreamClusterAsyncFlushFileStoreFlushOnSnapshot(t *testing.T) {
@@ -9154,11 +9181,10 @@ func TestJetStreamClusterAsyncFlushFileStoreFlushOnSnapshot(t *testing.T) {
 	defer nc.Close()
 
 	_, err := jsStreamCreate(t, nc, &StreamConfig{
-		Name:            "TEST",
-		Subjects:        []string{"foo"},
-		Storage:         FileStorage,
-		Replicas:        3,
-		AllowAsyncFlush: true,
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Storage:  FileStorage,
+		Replicas: 3,
 	})
 	require_NoError(t, err)
 
@@ -9230,6 +9256,9 @@ func TestJetStreamClusterAsyncFlushFileStoreFlushOnSnapshot(t *testing.T) {
 }
 
 func TestJetStreamClusterMsgGetReadAfterWrite(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9270,6 +9299,9 @@ func TestJetStreamClusterMsgGetReadAfterWrite(t *testing.T) {
 }
 
 func TestJetStreamClusterMsgGetMonotonicRead(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9306,6 +9338,9 @@ func TestJetStreamClusterMsgGetMonotonicRead(t *testing.T) {
 }
 
 func TestJetStreamClusterDirectGetReadAfterWrite(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9367,6 +9402,9 @@ func TestJetStreamClusterDirectGetReadAfterWrite(t *testing.T) {
 }
 
 func TestJetStreamClusterMirrorDirectGetReadAfterWrite(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9488,6 +9526,9 @@ func TestJetStreamClusterMirrorDirectGetReadAfterWrite(t *testing.T) {
 }
 
 func TestJetStreamClusterDirectGetReadAfterWriteOutdatedFollower(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	test := func(t *testing.T, templ string, accName string, authenticated bool) {
 		c := createJetStreamClusterWithTemplate(t, templ, "R3S", 3)
 		defer c.shutdown()
@@ -9591,6 +9632,9 @@ func TestJetStreamClusterDirectGetReadAfterWriteOutdatedFollower(t *testing.T) {
 }
 
 func TestJetStreamClusterDirectGetMonotonicRead(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9644,6 +9688,9 @@ func TestJetStreamClusterDirectGetMonotonicRead(t *testing.T) {
 }
 
 func TestJetStreamClusterDirectGetLastBySubjectReadAfterWrite(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9705,6 +9752,9 @@ func TestJetStreamClusterDirectGetLastBySubjectReadAfterWrite(t *testing.T) {
 }
 
 func TestJetStreamClusterDirectGetLastBySubjectReadAfterWriteOutdatedFollower(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9758,6 +9808,9 @@ func TestJetStreamClusterDirectGetLastBySubjectReadAfterWriteOutdatedFollower(t
 }
 
 func TestJetStreamClusterDirectGetLastBySubjectMonotonicRead(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -9808,6 +9861,9 @@ func TestJetStreamClusterDirectGetLastBySubjectMonotonicRead(t *testing.T) {
 }
 
 func TestJetStreamClusterConsumerReadAfterWrite(t *testing.T) {
+	// TODO(mvv): revisit for 2.14+
+	t.Skip()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 7ce6ff23..426a1b09 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -9867,3 +9867,377 @@ func TestJetStreamConsumerNotInactiveDuringAckWaitBackoff(t *testing.T) {
 	t.Run("R1", func(t *testing.T) { test(t, 1) })
 	t.Run("R3", func(t *testing.T) { test(t, 3) })
 }
+
+func TestSortingConsumerPullRequests(t *testing.T) {
+
+	for _, test := range []struct {
+		name     string
+		requests []waitingRequest
+		expected []waitingRequest
+	}{
+		{
+			name: "sort",
+			requests: []waitingRequest{
+				{priorityGroup: &PriorityGroup{Priority: 1}},
+				{priorityGroup: &PriorityGroup{Priority: 2}},
+				{priorityGroup: &PriorityGroup{Priority: 1}},
+				{priorityGroup: &PriorityGroup{Priority: 3}},
+			},
+			expected: []waitingRequest{
+				{priorityGroup: &PriorityGroup{Priority: 1}},
+				{priorityGroup: &PriorityGroup{Priority: 1}},
+				{priorityGroup: &PriorityGroup{Priority: 2}},
+				{priorityGroup: &PriorityGroup{Priority: 3}},
+			},
+		},
+		{
+			name: "test if sort is stable",
+			requests: []waitingRequest{
+				{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1a"},
+				{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2a"},
+				{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1b"},
+				{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2b"},
+				{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1c"},
+				{priorityGroup: &PriorityGroup{Priority: 3}, reply: "3a"},
+				{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2c"},
+			},
+			expected: []waitingRequest{
+				{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1a"},
+				{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1b"},
+				{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1c"},
+				{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2a"},
+				{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2b"},
+				{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2c"},
+				{priorityGroup: &PriorityGroup{Priority: 3}, reply: "3a"},
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			t.Helper()
+
+			wq := newWaitQueue(100)
+
+			for _, r := range test.requests {
+				err := wq.addPrioritized(&r)
+				require_NoError(t, err)
+			}
+
+			if wq.n != len(test.expected) {
+				t.Fatalf("Expected %d requests, got %d", len(test.expected), wq.n)
+			}
+
+			// Verify order
+			for i, expected := range test.expected {
+				current := wq.peek()
+				if current == nil {
+					t.Fatalf("Expected request at position %d, but queue was empty", i)
+				}
+				if current.priorityGroup.Priority != expected.priorityGroup.Priority {
+					t.Fatalf("At position %d: expected priority %d, got %d",
+						i, expected.priorityGroup.Priority, current.priorityGroup.Priority)
+				}
+				if current.reply != expected.reply {
+					t.Fatalf("At position %d: expected reply %q, got %q", i, expected.reply, current.reply)
+				}
+				wq.removeCurrent()
+			}
+
+		})
+	}
+}
+
+func TestWaitQueuePopAndRequeue(t *testing.T) {
+	t.Run("basic requeue with batches", func(t *testing.T) {
+		wq := newWaitQueue(100)
+
+		// Add elements with different priorities
+		reqs := []waitingRequest{
+			{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1a", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1b", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1c", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2a", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2b", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2c", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 3}, reply: "3a", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 3}, reply: "3b", n: 3},
+			{priorityGroup: &PriorityGroup{Priority: 3}, reply: "3c", n: 3},
+		}
+
+		for i := range reqs {
+			err := wq.addPrioritized(&reqs[i])
+			require_NoError(t, err)
+		}
+
+		i := 0
+		j := 0
+		for {
+			wr := wq.popAndRequeue()
+			require_Equal(t, wr.reply, fmt.Sprintf("%da", j+1))
+			wr = wq.popAndRequeue()
+			require_Equal(t, wr.reply, fmt.Sprintf("%db", j+1))
+			wr = wq.popAndRequeue()
+			require_Equal(t, wr.reply, fmt.Sprintf("%dc", j+1))
+
+			i++
+			if i%3 == 0 {
+				j++
+			}
+			require_Equal(t, wq.n, 9-(j*3))
+			if j == 2 {
+				break
+			}
+		}
+	})
+
+	t.Run("request removal when fully served", func(t *testing.T) {
+		wq := newWaitQueue(100)
+
+		reqs := []waitingRequest{
+			{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1a", n: 2}, // Will be removed after 2 pops
+			{priorityGroup: &PriorityGroup{Priority: 1}, reply: "1b", n: 1}, // Will be removed after 1 pop
+			{priorityGroup: &PriorityGroup{Priority: 2}, reply: "2a", n: 3}, // Will remain
+		}
+
+		for i := range reqs {
+			err := wq.addPrioritized(&reqs[i])
+			require_NoError(t, err)
+		}
+		initialCount := wq.n
+
+		// Pop 1a first time (n=2 -> n=1, should requeue)
+		wr := wq.popAndRequeue()
+		if wr == nil || wr.reply != "1a" || wr.n != 1 {
+			t.Fatalf("Expected 1a with n=1, got %v with n=%d", wr, wr.n)
+		}
+		if wq.n != initialCount {
+			t.Fatalf("Queue size should remain %d, got %d", initialCount, wq.n)
+		}
+
+		// Pop 1b (n=1 -> n=0, should be removed)
+		wr = wq.popAndRequeue()
+		if wr == nil || wr.reply != "1b" || wr.n != 0 {
+			t.Fatalf("Expected 1b with n=0, got %v with n=%d", wr, wr.n)
+		}
+		if wq.n != initialCount-1 {
+			t.Fatalf("Queue size should be %d after 1b removal, got %d", initialCount-1, wq.n)
+		}
+
+		// Pop 1a second time (n=1 -> n=0, should be removed)
+		wr = wq.popAndRequeue()
+		if wr == nil || wr.reply != "1a" || wr.n != 0 {
+			t.Fatalf("Expected 1a with n=0, got %v with n=%d", wr, wr.n)
+		}
+		if wq.n != initialCount-2 {
+			t.Fatalf("Queue size should be %d after 1a removal, got %d", initialCount-2, wq.n)
+		}
+
+		// Only 2a should remain
+		next := wq.peek()
+		if next == nil || next.reply != "2a" || next.n != 3 {
+			t.Fatalf("Expected only 2a with n=3 to remain, got %v with n=%d", next, next.n)
+		}
+	})
+
+	t.Run("single element lifecycle", func(t *testing.T) {
+		wq := newWaitQueue(10)
+		req := waitingRequest{priorityGroup: &PriorityGroup{Priority: 1}, reply: "single", n: 2}
+		wq.add(&req)
+
+		// First pop (n=2 -> n=1, should stay)
+		result := wq.popAndRequeue()
+		if result == nil || result.reply != "single" || result.n != 1 {
+			t.Fatalf("Expected single with n=1, got %v with n=%d", result, result.n)
+		}
+		if wq.n != 1 {
+			t.Fatalf("Queue should still have 1 element, got %d", wq.n)
+		}
+
+		// Second pop (n=1 -> n=0, should be removed)
+		result = wq.popAndRequeue()
+		if result == nil || result.reply != "single" || result.n != 0 {
+			t.Fatalf("Expected single with n=0, got %v with n=%d", result, result.n)
+		}
+		if wq.n != 0 {
+			t.Fatalf("Queue should be empty after removal, got %d elements", wq.n)
+		}
+		if wq.head != nil {
+			t.Fatalf("Queue head should be nil after removal, got %v", wq.head)
+		}
+	})
+}
+
+func TestJetStreamConsumerPrioritized(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)
+
+	acc, err := s.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err := acc.lookupStream("TEST")
+	require_NoError(t, err)
+
+	_, err = mset.addConsumer(&ConsumerConfig{
+		Durable:        "CONSUMER",
+		AckPolicy:      AckNone,
+		FilterSubject:  "foo",
+		PriorityGroups: []string{"A"},
+		PriorityPolicy: PriorityPrioritized,
+	})
+	require_NoError(t, err)
+
+	o := mset.lookupConsumer("CONSUMER")
+	require_NotNil(t, o)
+
+	// Helper function to send pull request and get response
+	sendPullRequest := func(t *testing.T, inbox string, priority int, batch int) *nats.Subscription {
+		sub, err := nc.SubscribeSync(inbox)
+		require_NoError(t, err)
+
+		req := JSApiConsumerGetNextRequest{
+			Batch:   batch,
+			Expires: 10 * time.Second,
+			PriorityGroup: PriorityGroup{
+				Group:    "A",
+				Priority: priority,
+			},
+		}
+		reqb, _ := json.Marshal(req)
+		err = nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.CONSUMER", inbox, reqb)
+		require_NoError(t, err)
+
+		return sub
+	}
+
+	t.Run("invalid priority number", func(t *testing.T) {
+
+		sub := sendPullRequest(t, "invalid_priority", 10, 1)
+
+		msg, err := sub.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_Equal(t, msg.Header.Get("Status"), "400")
+
+		sub2 := sendPullRequest(t, "invalid_priority", -5, 1)
+		require_Equal(t, msg.Header.Get("Status"), "400")
+
+		msg, err = sub2.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_Equal(t, msg.Header.Get("Status"), "400")
+
+	})
+
+	t.Run("priority ordering", func(t *testing.T) {
+		priority3 := sendPullRequest(t, "priority3", 3, 1) // Priority 3
+		priority1 := sendPullRequest(t, "priority1", 1, 1) // Priority 1 (should be served first)
+		priority2 := sendPullRequest(t, "priority2", 2, 2) // Priority 2
+
+		// Small delay to ensure requests are processed
+		time.Sleep(50 * time.Millisecond)
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "message"))
+		require_NoError(t, err)
+
+		// Priority 3 should time out
+		_, err := priority3.NextMsg(200 * time.Millisecond)
+		require_Error(t, err, nats.ErrTimeout)
+
+		// Priority 2 should time out
+		_, err = priority2.NextMsg(200 * time.Millisecond)
+		require_Error(t, err, nats.ErrTimeout)
+
+		// Priority 1 should get message first
+		msg, err := priority1.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "message"))
+		require_NoError(t, err)
+
+		// Priority 3 should time out.
+		_, err = priority3.NextMsg(200 * time.Millisecond)
+		require_Error(t, err, nats.ErrTimeout)
+
+		// Priority 2 should get message next
+		msg, err = priority2.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "message"))
+		require_NoError(t, err)
+
+		// Priority 2 should get message next
+		msg, err = priority2.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		// Priority 3 should time out
+		_, err = priority3.NextMsg(2 * time.Second)
+		require_Error(t, err, nats.ErrTimeout)
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "message"))
+		require_NoError(t, err)
+
+		// Priority 3 should get message next
+		msg, err = priority3.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+	})
+
+	t.Run("dynamic priority interruption", func(t *testing.T) {
+
+		inbox3 := nats.NewInbox()
+		sub3 := sendPullRequest(t, inbox3, 3, 3)
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "msg"))
+		require_NoError(t, err)
+
+		msg, err := sub3.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		// Now, despite priority 3 still having pending messages, a new pull request
+		// with a lower priority should be able to take over the delivery.
+		inbox2 := nats.NewInbox()
+		sub2 := sendPullRequest(t, inbox2, 2, 2)
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "msg"))
+		require_NoError(t, err)
+
+		msg, err = sub2.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		// The same should happen with priority 1.
+		inbox1 := nats.NewInbox()
+		sub1 := sendPullRequest(t, inbox1, 1, 1) // Priority 1, batch 3
+
+		_, err = js.Publish("foo", fmt.Appendf(nil, "msg"))
+		require_NoError(t, err)
+		_, err = js.Publish("foo", fmt.Appendf(nil, "msg"))
+		require_NoError(t, err)
+		_, err = js.Publish("foo", fmt.Appendf(nil, "msg"))
+		require_NoError(t, err)
+
+		msg, err = sub1.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		// Now priority 2 should take over for last message.
+		msg, err = sub2.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+
+		// Finally, priority 3 should get its last message.
+		msg, err = sub3.NextMsg(2 * time.Second)
+		require_NoError(t, err)
+		require_NotNil(t, msg)
+	})
+}
diff --git a/server/jetstream_versioning_test.go b/server/jetstream_versioning_test.go
index 3beb9f0a..2eed068c 100644
--- a/server/jetstream_versioning_test.go
+++ b/server/jetstream_versioning_test.go
@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"math"
 	"reflect"
 	"strconv"
 	"testing"
@@ -82,11 +83,6 @@ func TestJetStreamSetStaticStreamMetadata(t *testing.T) {
 			cfg:              &StreamConfig{AllowAtomicPublish: true},
 			expectedMetadata: metadataAtLevel("2"),
 		},
-		{
-			desc:             "AllowAsyncFlush",
-			cfg:              &StreamConfig{AllowAsyncFlush: true},
-			expectedMetadata: metadataAtLevel("2"),
-		},
 	} {
 		t.Run(test.desc, func(t *testing.T) {
 			setStaticStreamMetadata(test.cfg)
@@ -229,11 +225,6 @@ func TestJetStreamSetStaticConsumerMetadata(t *testing.T) {
 			cfg:              &ConsumerConfig{PriorityPolicy: PriorityPinnedClient, PriorityGroups: []string{"a"}},
 			expectedMetadata: metadataAtLevel("1"),
 		},
-		{
-			desc:             "MinLastSeq",
-			cfg:              &ConsumerConfig{MinLastSeq: 1},
-			expectedMetadata: metadataAtLevel("2"),
-		},
 	} {
 		t.Run(test.desc, func(t *testing.T) {
 			setStaticConsumerMetadata(test.cfg)
@@ -623,3 +614,80 @@ func restoreEmptyStream(t *testing.T, nc *nats.Conn, replicas int) {
 	require_NoError(t, err)
 	require_True(t, reflect.DeepEqual(cresp.Config.Metadata, expectedMetadata))
 }
+
+func TestJetStreamApiErrorOnRequiredApiLevel(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	var subs []*subscription
+	s.getJetStream().apiSubs.All(&subs)
+	require_True(t, len(subs) > 0)
+	for _, sub := range subs {
+		apiSubject := string(sub.subject)
+		t.Run(apiSubject, func(t *testing.T) {
+			req := nats.NewMsg(apiSubject)
+			req.Header.Set("Nats-Required-Api-Level", strconv.Itoa(math.MaxInt))
+			msg, err := nc.RequestMsg(req, time.Second)
+			require_NoError(t, err)
+
+			var resp ApiResponse
+			require_NoError(t, json.Unmarshal(msg.Data, &resp))
+			require_True(t, resp.Error != nil)
+			require_Error(t, resp.Error, NewJSRequiredApiLevelError())
+		})
+	}
+}
+
+func TestJetStreamApiErrorOnRequiredApiLevelDirectGet(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{"foo"},
+		AllowDirect: true,
+	})
+	require_NoError(t, err)
+
+	req := nats.NewMsg(fmt.Sprintf(JSDirectMsgGetT, "TEST"))
+	req.Header.Set("Nats-Required-Api-Level", strconv.Itoa(math.MaxInt))
+	msg, err := nc.RequestMsg(req, time.Second)
+	require_NoError(t, err)
+	require_Equal(t, msg.Header.Get("Status"), "412")
+	require_Equal(t, msg.Header.Get("Description"), "Required Api Level")
+
+	req = nats.NewMsg(fmt.Sprintf(JSDirectGetLastBySubjectT, "TEST", "foo"))
+	req.Header.Set("Nats-Required-Api-Level", strconv.Itoa(math.MaxInt))
+	msg, err = nc.RequestMsg(req, time.Second)
+	require_NoError(t, err)
+	require_Equal(t, msg.Header.Get("Status"), "412")
+	require_Equal(t, msg.Header.Get("Description"), "Required Api Level")
+}
+
+func TestJetStreamApiErrorOnRequiredApiLevelPullConsumerNextMsg(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{"foo"},
+		AllowDirect: true,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "CONSUMER"})
+	require_NoError(t, err)
+
+	req := nats.NewMsg(fmt.Sprintf(JSApiRequestNextT, "TEST", "CONSUMER"))
+	req.Header.Set("Nats-Required-Api-Level", strconv.Itoa(math.MaxInt))
+	msg, err := nc.RequestMsg(req, time.Second)
+	require_NoError(t, err)
+	require_Equal(t, msg.Header.Get("Status"), "412")
+	require_Equal(t, msg.Header.Get("Description"), "Required Api Level")
+}
diff --git a/server/opts_test.go b/server/opts_test.go
index ffb40a6a..4f47cbaa 100644
--- a/server/opts_test.go
+++ b/server/opts_test.go
@@ -3753,3 +3753,171 @@ func parseConfigTolerantly(t *testing.T, data string) (*Options, error) {
 
 	return o, nil
 }
+
+func TestOptionsProxyTrustedKeys(t *testing.T) {
+	o := DefaultOptions()
+	o.Proxies = &ProxiesConfig{
+		Trusted: []*ProxyConfig{
+			{Key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"},
+			{Key: "bad1"},
+			{Key: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"},
+			{Key: "bad2"},
+		},
+	}
+	err := validateOptions(o)
+	require_Error(t, err)
+	require_Equal(t, "proxy trusted key \"bad1\" is invalid", err.Error())
+
+	o.Proxies = &ProxiesConfig{
+		Trusted: []*ProxyConfig{
+			{Key: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"},
+			{Key: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"},
+		},
+	}
+	s := RunServer(o)
+	defer s.Shutdown()
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	opts := s.getOpts()
+	for i, kp := range s.proxiesKeyPairs {
+		pub, err := kp.PublicKey()
+		require_NoError(t, err)
+		require_Equal(t, opts.Proxies.Trusted[i].Key, pub)
+	}
+}
+
+func TestOptionsProxyRequired(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		authorization {
+			user: user
+			password: pwd
+			proxy_required: true
+		}
+	`))
+	o, err := ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Equal(t, o.Username, "user")
+	require_Equal(t, o.Password, "pwd")
+	require_True(t, o.ProxyRequired)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		authorization {
+			users: [
+				{user: user1, password: pwd1}
+				{user: user2, password: pwd2, proxy_required: true}
+				{user: user3, password: pwd3, proxy_required: false}
+				{nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3", proxy_required: true}
+				{nkey: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI", proxy_required: false}
+			]
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+
+	checkUsersAndNkeys := func(users []*User, hasNkeys bool, nkeys []*NkeyUser) {
+		t.Helper()
+		var found bool
+
+		require_Len(t, len(users), 3)
+		for _, u := range users {
+			switch u.Username {
+			case "user1", "user3":
+				require_False(t, u.ProxyRequired)
+			case "user2":
+				require_True(t, u.ProxyRequired)
+				found = true
+			}
+		}
+		require_True(t, found)
+
+		if !hasNkeys {
+			return
+		}
+
+		found = false
+		require_Len(t, len(nkeys), 2)
+		for _, u := range nkeys {
+			switch u.Nkey {
+			case "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3":
+				require_True(t, u.ProxyRequired)
+				found = true
+			case "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI":
+				require_False(t, u.ProxyRequired)
+			}
+		}
+		require_True(t, found)
+	}
+	checkUsersAndNkeys(o.Users, true, o.Nkeys)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		accounts {
+			A {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+					{nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3", proxy_required: true}
+					{nkey: "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI", proxy_required: false}
+				]
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Len(t, len(o.Accounts), 1)
+	require_Equal(t, o.Accounts[0].Name, "A")
+	checkUsersAndNkeys(o.Users, true, o.Nkeys)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				user: user
+				password: pwd
+				proxy_required: true
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Equal(t, o.LeafNode.Username, "user")
+	require_Equal(t, o.LeafNode.Password, "pwd")
+	require_True(t, o.LeafNode.ProxyRequired)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				nkey: "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3"
+				proxy_required: true
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	require_Equal(t, o.LeafNode.Nkey, "UCARKS2E3KVB7YORL2DG34XLT7PUCOL2SVM7YXV6ETHLW6Z46UUJ2VZ3")
+	require_True(t, o.LeafNode.ProxyRequired)
+
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes: {
+			port: -1
+			authorization {
+				users: [
+					{user: user1, password: pwd1}
+					{user: user2, password: pwd2, proxy_required: true}
+					{user: user3, password: pwd3, proxy_required: false}
+				]
+			}
+		}
+	`))
+	o, err = ProcessConfigFile(conf)
+	require_NoError(t, err)
+	checkUsersAndNkeys(o.LeafNode.Users, false, nil)
+}
diff --git a/server/raft_test.go b/server/raft_test.go
index d6a9359b..84fcf1a5 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -3158,6 +3158,101 @@ func TestNRGStepdownWithHighestTermDuringCatchup(t *testing.T) {
 	require_Equal(t, n.pindex, 2)
 }
 
+func TestNRGTruncateOnStartup(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	s := c.servers[0] // RunBasicJetStreamServer not available
+	defer c.shutdown()
+
+	storeDir := t.TempDir()
+	fcfg := FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMediumBlockSize, AsyncFlush: false, srv: s}
+	scfg := StreamConfig{Name: "RAFT", Storage: FileStorage}
+	fs, err := newFileStore(fcfg, scfg)
+	require_NoError(t, err)
+
+	cfg := &RaftConfig{Name: "TEST", Store: storeDir, Log: fs}
+
+	err = s.bootstrapRaftNode(cfg, nil, false)
+	require_NoError(t, err)
+	n, err := s.initRaftNode(globalAccountName, cfg, pprofLabels{})
+	require_NoError(t, err)
+
+	// Create a sample entry, the content doesn't matter, just that it's stored.
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+
+	nats0 := "S1Nunr6R" // "nats-0"
+	aeMsg1 := encode(t, &appendEntry{leader: nats0, term: 1, commit: 0, pterm: 0, pindex: 0, entries: entries})
+	aeMsg2 := encode(t, &appendEntry{leader: nats0, term: 1, commit: 0, pterm: 1, pindex: 1, entries: entries})
+
+	// Store two messages the normal way.
+	n.processAppendEntry(aeMsg1, n.aesub)
+	n.processAppendEntry(aeMsg2, n.aesub)
+	require_Equal(t, n.pindex, 2)
+
+	state := n.wal.State()
+	require_Equal(t, state.Msgs, 2)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 2)
+	require_Equal(t, state.NumDeleted, 0)
+
+	// Simulate a truncation that's only performed halfway, and we got hard killed at some point.
+	removed, err := n.wal.RemoveMsg(2)
+	require_True(t, removed)
+	require_NoError(t, err)
+
+	state = n.wal.State()
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 2)
+	require_Equal(t, state.NumDeleted, 1)
+
+	// Restart.
+	n.Stop()
+	n.WaitForStop()
+	require_NoError(t, fs.Stop())
+	fs, err = newFileStore(fcfg, scfg)
+	require_NoError(t, err)
+	cfg = &RaftConfig{Name: "TEST", Store: storeDir, Log: fs}
+	n, err = s.initRaftNode(globalAccountName, cfg, pprofLabels{})
+	require_NoError(t, err)
+
+	// Should truncate the WAL on startup, the message was removed.
+	state = n.wal.State()
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 1)
+	require_Equal(t, state.NumDeleted, 0)
+
+	// Store an invalid append entry manually.
+	aeMsg2.pindex = 0
+	aeMsg2 = encode(t, aeMsg2)
+	_, _, err = n.wal.StoreMsg(_EMPTY_, nil, aeMsg2.buf, 0)
+	require_NoError(t, err)
+
+	state = n.wal.State()
+	require_Equal(t, state.Msgs, 2)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 2)
+	require_Equal(t, state.NumDeleted, 0)
+
+	// Restart.
+	n.Stop()
+	n.WaitForStop()
+	require_NoError(t, fs.Stop())
+	fs, err = newFileStore(fcfg, scfg)
+	require_NoError(t, err)
+	cfg = &RaftConfig{Name: "TEST", Store: storeDir, Log: fs}
+	n, err = s.initRaftNode(globalAccountName, cfg, pprofLabels{})
+	require_NoError(t, err)
+
+	// Should truncate the WAL on startup, the append entry is invalid.
+	state = n.wal.State()
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 1)
+	require_Equal(t, state.NumDeleted, 0)
+}
+
 // This is a RaftChainOfBlocks test where a block is proposed and then we wait for all replicas to apply it before
 // proposing the next one.
 // The test may fail if:
diff --git a/test/client_auth_test.go b/test/client_auth_test.go
index 90926bd3..51e8cb02 100644
--- a/test/client_auth_test.go
+++ b/test/client_auth_test.go
@@ -14,13 +14,16 @@
 package test
 
 import (
+	"encoding/base64"
 	"fmt"
+	"net"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 )
 
 func TestMultipleUserAuth(t *testing.T) {
@@ -156,3 +159,357 @@ func TestClientConnectInfo(t *testing.T) {
 		})
 	}
 }
+
+func TestProxyKeyVerification(t *testing.T) {
+	u1, _ := nkeys.CreateUser()
+	u1Pub, _ := u1.PublicKey()
+
+	u2, _ := nkeys.CreateUser()
+	u2Pub, _ := u2.PublicKey()
+
+	u3, _ := nkeys.CreateUser()
+	u3Pub, _ := u3.PublicKey()
+
+	tmpl := `
+		listen: "127.0.0.1:-1"
+		proxies {
+			trusted [
+				{key: %q}
+				{key: %q}
+			]
+		}
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+	`
+	conf := createConfFile(t, fmt.Appendf(nil, tmpl, u1Pub, u2Pub))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	varz, err := s.Varz(nil)
+	if err != nil {
+		t.Fatalf("Error getting varz: %v", err)
+	}
+	vproxies := varz.Proxies
+	if vproxies == nil {
+		t.Fatal("Expected proxies to be not nil")
+	}
+	trusted := vproxies.Trusted
+	if len(trusted) != 2 {
+		t.Fatalf("Expected 2 trusted proxies, got %+v", trusted)
+	}
+	if k := trusted[0].Key; k != u1Pub {
+		t.Fatalf("Expected key to be %q, got %q", u1Pub, k)
+	}
+	if k := trusted[1].Key; k != u2Pub {
+		t.Fatalf("Expected key to be %q, got %q", u2Pub, k)
+	}
+
+	var currentCID uint64
+	connect := func(t *testing.T, kp nkeys.KeyPair, bsig bool, port int, isClient bool, user string) (expectFun, net.Conn) {
+		c := createClientConn(t, "127.0.0.1", port)
+		expect := expectCommand(t, c)
+		info := checkInfoMsg(t, c)
+		currentCID = info.CID
+		if info.Nonce == "" {
+			c.Close()
+			t.Fatalf("Expected nonce from INFO, got %+v", info)
+		}
+		if info.JSApiLevel != server.JSApiLevel {
+			c.Close()
+			t.Fatalf("Expected JSApiLevel to be set to %v, got %v", server.JSApiLevel, info.JSApiLevel)
+		}
+		sig, err := kp.Sign([]byte(info.Nonce))
+		if err != nil {
+			c.Close()
+			t.Fatalf("Error signing nonce %q: %v", info.Nonce, err)
+		}
+		encodedSig := make([]byte, base64.RawURLEncoding.EncodedLen(len(sig)))
+		base64.RawURLEncoding.Encode(encodedSig, sig)
+		if bsig {
+			encodedSig[0] = '*'
+		}
+		sendProto(t, c,
+			fmt.Sprintf("CONNECT {\"pedantic\":false,\"verbose\":false,\"user\":%q,\"pass\":\"pwd\",\"proxy_sig\":\"%s\"}\r\n",
+				user, string(encodedSig)))
+		if isClient {
+			sendProto(t, c, "PING\r\n")
+		}
+		return expect, c
+	}
+
+	checkClosedState := func(t *testing.T, cid uint64, reason server.ClosedState) {
+		// Closed connection getting in the closed state is done in a go routine
+		// so make sure we wait for it.
+		checkFor(t, time.Second, 100*time.Millisecond, func() error {
+			connz, err := s.Connz(&server.ConnzOptions{
+				CID:   cid,
+				State: server.ConnClosed,
+			})
+			if err != nil {
+				return fmt.Errorf("Error getting connz for closed connection %v: %v", cid, err)
+			}
+			if len(connz.Conns) != 1 {
+				return fmt.Errorf("Expected 1 connection, got %v", connz.Conns)
+			}
+			conn := connz.Conns[0]
+			if conn.Reason != reason.String() {
+				return fmt.Errorf("Expected reason %q, got %q", reason.String(), conn.Reason)
+			}
+			return nil
+		})
+	}
+
+	for _, test := range []struct {
+		name string
+		kp   nkeys.KeyPair
+		bsig bool
+		ok   bool
+	}{
+		{"key present first", u1, false, true},
+		{"key present last", u2, false, true},
+		{"key not present", u3, false, false},
+		{"bad signature", u2, true, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			for _, subTest := range []struct {
+				name     string
+				port     int
+				isClient bool
+			}{
+				{"client", o.Port, true},
+				{"leafnodes", o.LeafNode.Port, false},
+			} {
+				t.Run(subTest.name, func(t *testing.T) {
+					expect, c := connect(t, test.kp, test.bsig, subTest.port, subTest.isClient, "user")
+					defer c.Close()
+					if test.ok {
+						var p *server.ProxyInfo
+						if subTest.isClient {
+							expect(pongRe)
+
+							connz, err := s.Connz(nil)
+							if err != nil {
+								t.Fatalf("Error getting connz: %v", err)
+							}
+							if n := len(connz.Conns); n != 1 {
+								t.Fatalf("Expected 1 ConnInfo, got %v", n)
+							}
+							ci := connz.Conns[0]
+							p = ci.Proxy
+						} else {
+							expect(infoRe)
+
+							leafz, err := s.Leafz(nil)
+							if err != nil {
+								t.Fatalf("Error getting leafz: %v", err)
+							}
+							if n := len(leafz.Leafs); n != 1 {
+								t.Fatalf("Expected 1 LeafInfo, got %v", n)
+							}
+							li := leafz.Leafs[0]
+							if li.Proxy == nil {
+								t.Fatalf("Required Proxy to be present, was not: %+v", li)
+							}
+							p = li.Proxy
+						}
+						if p == nil {
+							t.Fatalf("Required Proxy to be present, was not")
+						}
+						pub, _ := test.kp.PublicKey()
+						if p.Key != pub {
+							t.Fatalf("Expected public key to be %q, got %q", pub, p.Key)
+						}
+					} else {
+						expect(errRe)
+						checkClosedState(t, currentCID, server.ProxyNotTrusted)
+					}
+				})
+			}
+		})
+	}
+
+	// Create a connection using u1Pub and we will config reload and remove
+	// that key from the list. The client should get disconnected.
+	expectc, c := connect(t, u1, false, o.Port, true, "user")
+	defer c.Close()
+	expectc(pongRe)
+	// Capture this connection CID before the next connect.
+	cid1 := currentCID
+
+	expectl, l := connect(t, u1, false, o.LeafNode.Port, false, "user")
+	defer l.Close()
+	expectl(infoRe)
+	cid2 := currentCID
+	checkLeafNodeConnected(t, s)
+
+	os.WriteFile(conf, fmt.Appendf(nil, tmpl, u3Pub, u2Pub), 0660)
+	if err := s.Reload(); err != nil {
+		t.Fatalf("Reload failed: %v", err)
+	}
+	// Connections should get disconnected.
+	// We need to consume what is sent by the server, but for leaf we may
+	// get some LS+, etc... so just consumer until we get the io.EOF
+	readAll := func(c net.Conn) {
+		for {
+			var buf [1024]byte
+			if _, err := c.Read(buf[:]); err != nil {
+				break
+			}
+		}
+	}
+	readAll(c)
+	expectDisconnect(t, c)
+	readAll(l)
+	expectDisconnect(t, l)
+	c.Close()
+	l.Close()
+
+	// Now check that the connection with CID `cid` was closed because the
+	// proxy is (no longer) trusted.
+	checkClosedState(t, cid1, server.ProxyNotTrusted)
+	checkClosedState(t, cid2, server.ProxyNotTrusted)
+
+	varz, err = s.Varz(nil)
+	if err != nil {
+		t.Fatalf("Error getting varz: %v", err)
+	}
+	vproxies = varz.Proxies
+	if vproxies == nil {
+		t.Fatal("Expected proxies to be not nil")
+	}
+	trusted = vproxies.Trusted
+	if len(trusted) != 2 {
+		t.Fatalf("Expected 2 trusted proxies, got %+v", trusted)
+	}
+	if k := trusted[0].Key; k != u3Pub {
+		t.Fatalf("Expected key to be %q, got %q", u3Pub, k)
+	}
+	if k := trusted[1].Key; k != u2Pub {
+		t.Fatalf("Expected key to be %q, got %q", u2Pub, k)
+	}
+
+	// And we should be able to connect using u3 key pair.
+	expectc, c = connect(t, u3, false, o.Port, true, "user")
+	expectc(pongRe)
+	c.Close()
+	expectl, l = connect(t, u3, false, o.LeafNode.Port, false, "user")
+	expectl(infoRe)
+	l.Close()
+
+	// But now using u1 key pair should fail.
+	expectc, c = connect(t, u1, false, o.Port, true, "user")
+	expectc(errRe)
+	c.Close()
+	expectl, l = connect(t, u1, false, o.LeafNode.Port, false, "user")
+	expectl(errRe)
+	l.Close()
+
+	s.Shutdown()
+
+	// Now check that if no trusted proxy are provided, a proxy connection is
+	// accepted. Use leafnodes since we always send NONCE for this type of connection.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	for _, test := range []struct {
+		name string
+		bsig bool
+	}{
+		{"no trusted proxy and correct signature", false},
+		{"no trusted proxy and invalid signature", true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			expect, c := connect(t, u1, test.bsig, o.LeafNode.Port, false, "user")
+			defer c.Close()
+			// Does not matter if bad signature, since there is no trusted keys
+			// in the server, we accept without signature check.
+			expect(infoRe)
+		})
+	}
+
+	s.Shutdown()
+
+	// Finally, check that if there are no configured trusted proxies
+	// and a user requires a proxy connection, then the connection will fail.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				users [
+					{user: u1, password: pwd}
+					{user: u2, password: pwd, proxy_required: true}
+				]
+			}
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	for _, test := range []struct {
+		name string
+		user string
+		ok   bool
+	}{
+		{"no trusted proxy configured no proxy required", "u1", true},
+		{"no trusted proxy configured and proxy required", "u2", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			// We can use any key pair here, it is not important.
+			expect, c := connect(t, u1, false, o.LeafNode.Port, false, test.user)
+			defer c.Close()
+			if test.ok {
+				expect(infoRe)
+			} else {
+				expect(errRe)
+				// Check that the reason is "proxy required"
+				checkClosedState(t, currentCID, server.ProxyRequired)
+			}
+		})
+	}
+
+	// Same test, but the "proxy_required" is at the authorization{} to-level,
+	// so any user should fail.
+	conf = createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			authorization {
+				users [
+					{user: u1, password: pwd}
+					{user: u2, password: pwd}
+				]
+				proxy_required: true
+			}
+		}
+	`))
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	t.Run("all users required to use trusted proxy", func(t *testing.T) {
+		for _, user := range []string{"u1", "u2"} {
+			// We can use any key pair here, it is not important.
+			expect, c := connect(t, u1, false, o.LeafNode.Port, false, user)
+			defer c.Close()
+			expect(errRe)
+			// Check that the reason is "proxy required"
+			checkClosedState(t, currentCID, server.ProxyRequired)
+		}
+	})
+
+	varz, err = s.Varz(nil)
+	if err != nil {
+		t.Fatalf("Error getting varz: %v", err)
+	}
+	vproxies = varz.Proxies
+	if vproxies != nil {
+		t.Fatalf("Expected proxies to be nil, got %+v", vproxies)
+	}
+}