Repository: nats-io/nats-server
Pull request #3409 (2022-08-27 20:54:56)

[IMPROVED] Secure consumer create
New consumer create that allows elevation of the stream and consumer names, and optional filter subject to the request subject.

Similar to changes in direct get allows proper security if needed for filter subject selection.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core


aricart: So this means that if you want to create a durable, the client specifies `durable_name`, otherwise `name`, or empty `name` and `durable_name` for an ephemeral
derekcollison: > So this means that if you want to create a durable, the client specifies `durable_name`, otherwise `name`, or empty `name` and `durable_name` for an ephemeral

You can also create a named ephemeral by specifying name and inactive threshold.

ripienaar: So feature wise we have still Durable and Ephemeral. 

Both will survive reboots
Both have a idle timeout
Both have a name
Both supports the user giving a name

Is it then that the only difference is essentially just positioning - a durable without a timeout is ephemeral? 

Then I guess implementation wise everything is a durable, the server only knows about durable consumers.  If a client doesnt set `durable_name` or `inactive_timeout` we supply them in line with ephemeral behaviors of today. The result is a durable with a timeout and a generated name.

You can additionally have R1 durables, memory only durables and so forth.  But all are durables.

Keeping the structures as is seem acceptible to me, no client changes needed to create an ephemeral or durable - the API simplification can use the features as it wish any old code continue to work?

So if above is not correct, what is the technical need for adding/deprecating around names, what can not be implemented without deprecating durable name? 

These kind of API key changes, not on a major version boundary is really disruptive, so much so that I am thinking of just abandoning the terraform provider if the server simply cant keep stable interfaces since those systems require that.

Patch:
diff --git a/server/const.go b/server/const.go
index b97c980c..cc132f02 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.9.0-RC.14"
+	VERSION = "2.9.0-RC.11"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 30c65cc9..70198d84 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -48,7 +48,9 @@ type ConsumerInfo struct {
 }
 
 type ConsumerConfig struct {
+	// Durable is deprecated. All consumers will have names. picked by clients.
 	Durable         string          `json:"durable_name,omitempty"`
+	Name            string          `json:"name,omitempty"`
 	Description     string          `json:"description,omitempty"`
 	DeliverPolicy   DeliverPolicy   `json:"deliver_policy"`
 	OptStartSeq     uint64          `json:"opt_start_seq,omitempty"`
@@ -353,10 +355,6 @@ func setConsumerConfigDefaults(config *ConsumerConfig, lim *JSLimitOpts, accLim
 	}
 }
 
-func (mset *stream) addConsumer(config *ConsumerConfig) (*consumer, error) {
-	return mset.addConsumerWithAssignment(config, _EMPTY_, nil, false)
-}
-
 // Check the consumer config. If we are recovering don't check filter subjects.
 func checkConsumerCfg(
 	config *ConsumerConfig,
@@ -530,6 +528,10 @@ func checkConsumerCfg(
 	return nil
 }
 
+func (mset *stream) addConsumer(config *ConsumerConfig) (*consumer, error) {
+	return mset.addConsumerWithAssignment(config, _EMPTY_, nil, false)
+}
+
 func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname string, ca *consumerAssignment, isRecovering bool) (*consumer, error) {
 	mset.mu.RLock()
 	s, jsa, tierName, cfg, acc := mset.srv, mset.jsa, mset.tier, mset.cfg, mset.acc
@@ -589,8 +591,14 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	}
 
 	// If this one is durable and already exists, we let that be ok as long as only updating what should be allowed.
+	var cName string
 	if isDurableConsumer(config) {
-		if eo, ok := mset.consumers[config.Durable]; ok {
+		cName = config.Durable
+	} else if config.Name != _EMPTY_ {
+		cName = config.Name
+	}
+	if cName != _EMPTY_ {
+		if eo, ok := mset.consumers[cName]; ok {
 			mset.mu.Unlock()
 			err := eo.updateConfig(config)
 			if err == nil {
@@ -674,11 +682,17 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	} else if oname != _EMPTY_ {
 		o.name = oname
 	} else {
-		for {
-			o.name = createConsumerName()
-			if _, ok := mset.consumers[o.name]; !ok {
-				break
+		if config.Name != _EMPTY_ {
+			o.name = config.Name
+		} else {
+			// Legacy ephemeral auto-generated.
+			for {
+				o.name = createConsumerName()
+				if _, ok := mset.consumers[o.name]; !ok {
+					break
+				}
 			}
+			config.Name = o.name
 		}
 	}
 	// Create ackMsgs queue now that we have a consumer name
@@ -768,6 +782,8 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		// Ephemerals will always have inactive thresholds.
 		// Add in 1 sec of jitter above and beyond the default of 5s.
 		o.dthresh = JsDeleteWaitTimeDefault + time.Duration(rand.Int63n(1000))*time.Millisecond
+		// Only stamp config with default sans jitter.
+		o.cfg.InactiveThreshold = JsDeleteWaitTimeDefault
 	}
 
 	if o.isPushMode() {
diff --git a/server/errors.json b/server/errors.json
index fc5f9327..cf464f62 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1288,5 +1288,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerCreateFilterSubjectMismatchErr",
+    "code": 400,
+    "error_code": 10131,
+    "description": "Consumer create request did not match filtered subject from create subject",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
-]
\ No newline at end of file
+]
diff --git a/server/filestore.go b/server/filestore.go
index 925ed368..d2bea361 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1235,8 +1235,6 @@ func (fs *fileStore) expireMsgsOnRecover() {
 				// Do this part by hand since not deleting one by one.
 				mb.first.seq, mb.first.ts = mb.last.seq+1, 0
 				mb.closeAndKeepIndex()
-				// Clear any global subject state.
-				fs.psim = make(map[string]*psi)
 			} else {
 				mb.dirtyCloseWithRemove(true)
 				deleted++
@@ -1848,10 +1846,11 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	// Race detector wants these protected.
 	mb.mu.Lock()
 	mb.llts, mb.lwts = ts, ts
+	mb.mu.Unlock()
+
 	// Remember our last sequence number.
 	mb.first.seq = fs.state.LastSeq + 1
 	mb.last.seq = fs.state.LastSeq
-	mb.mu.Unlock()
 
 	// If we know we will need this so go ahead and spin up.
 	if !fs.fip {
@@ -3066,17 +3065,12 @@ func (fs *fileStore) checkMsgs() *LostStreamData {
 
 	fs.checkAndFlushAllBlocks()
 
-	// Clear any global subject state.
-	fs.psim = make(map[string]*psi)
-
 	for _, mb := range fs.blks {
 		if ld, err := mb.rebuildState(); err != nil && ld != nil {
 			// Rebuild fs state too.
 			mb.fs.rebuildStateLocked(ld)
 		}
-		fs.populateGlobalPerSubjectInfo(mb)
 	}
-
 	return fs.ld
 }
 
@@ -3182,20 +3176,22 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 	// Decide if we write index info if flushing in place.
 	writeIndex := ts-mb.lwits > wiThresh
 
+	// Accounting
+	mb.updateAccounting(seq, ts, rl)
+
 	// Check if we are tracking per subject for our simple state.
-	if len(subj) > 0 && !mb.noTrack {
+	if len(subj) > 0 {
 		mb.ensurePerSubjectInfoLoaded()
-		if ss := mb.fss[subj]; ss != nil {
-			ss.Msgs++
-			ss.Last = seq
-		} else {
-			mb.fss[subj] = &SimpleState{Msgs: 1, First: seq, Last: seq}
+		if mb.fss != nil {
+			if ss := mb.fss[subj]; ss != nil {
+				ss.Msgs++
+				ss.Last = seq
+			} else {
+				mb.fss[subj] = &SimpleState{Msgs: 1, First: seq, Last: seq}
+			}
 		}
 	}
 
-	// Accounting
-	mb.updateAccounting(seq, ts, rl)
-
 	fch, werr := mb.fch, mb.werr
 
 	// If we should be flushing, or had a write error, do so here.
@@ -4994,12 +4990,6 @@ func (mb *msgBlock) closeAndKeepIndex() {
 	// Close
 	mb.dirtyCloseWithRemove(false)
 
-	// Make sure to remove fss state.
-	mb.fss = nil
-	if mb.sfn != _EMPTY_ {
-		os.Remove(mb.sfn)
-	}
-
 	// If we are encrypted we should reset our bek counter.
 	if mb.bek != nil {
 		if bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce); err == nil {
@@ -5194,19 +5184,13 @@ func (mb *msgBlock) loadPerSubjectInfo() ([]byte, error) {
 // Helper to make sure fss loaded if we are tracking.
 // Lock should be held
 func (mb *msgBlock) ensurePerSubjectInfoLoaded() error {
-	if mb.fss != nil || mb.noTrack {
-		return nil
-	}
-	if mb.msgs == 0 {
-		mb.fss = make(map[string]*SimpleState)
-		return nil
+	if mb.fss == nil && !mb.noTrack {
+		return mb.readPerSubjectInfo(true)
 	}
-	// Load from file.
-	return mb.readPerSubjectInfo(true)
+	return nil
 }
 
 // Called on recovery to populate the global psim state.
-// Lock should be held.
 func (fs *fileStore) populateGlobalPerSubjectInfo(mb *msgBlock) {
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
@@ -5214,20 +5198,6 @@ func (fs *fileStore) populateGlobalPerSubjectInfo(mb *msgBlock) {
 	if err := mb.readPerSubjectInfo(true); err != nil {
 		return
 	}
-
-	// Quick sanity check.
-	// TODO(dlc) - This is here to auto-clear a bug.
-	fssMsgs := uint64(0)
-	for subj, ss := range mb.fss {
-		if len(subj) > 0 {
-			fssMsgs += ss.Msgs
-		}
-	}
-	// If we are off rebuild.
-	if fssMsgs != mb.msgs {
-		mb.generatePerSubjectInfo(true)
-	}
-
 	// Now populate psim.
 	for subj, ss := range mb.fss {
 		if len(subj) > 0 {
diff --git a/server/gateway.go b/server/gateway.go
index 4cab3303..2a7de885 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -164,13 +164,6 @@ type srvGateway struct {
 
 	// This is to track recent subscriptions for a given connection
 	rsubs sync.Map
-	// This client will be used for SYSTEM clients when storing/looking up
-	// recent subscriptions in rsubs. This is because some code may not
-	// use the same actual *client object for SYSTEM client. For instance
-	// a raft node creates an internal client, that would be used to store
-	// the subscription in rsubs, but the sending part that checks for rsubs
-	// is using an internal client created in sendq.go's internalLoop.
-	sysCli *client
 
 	resolver  netResolver   // Used to resolve host name before calling net.Dial()
 	sqbsz     int           // Max buffer size to send queue subs protocol. Used for testing.
@@ -366,7 +359,6 @@ func (s *Server) newGateway(opts *Options) error {
 		resolver: opts.Gateway.resolver,
 		runknown: opts.Gateway.RejectUnknown,
 		oldHash:  getOldHash(opts.Gateway.Name),
-		sysCli:   &client{},
 	}
 	gateway.Lock()
 	defer gateway.Unlock()
@@ -2368,9 +2360,6 @@ func (s *Server) gatewayUpdateSubInterest(accName string, sub *subscription, cha
 	if sub.client != nil {
 		rsubs := &s.gateway.rsubs
 		c := sub.client
-		if c.kind == SYSTEM {
-			c = s.gateway.sysCli
-		}
 		sli, _ := rsubs.Load(c)
 		if change > 0 {
 			var sl *Sublist
@@ -2436,8 +2425,6 @@ func (g *srvGateway) shouldMapReplyForGatewaySend(c *client, acc *Account, reply
 		acc.mu.Lock()
 		c = acc.internalClient()
 		acc.mu.Unlock()
-	} else if c.kind == SYSTEM {
-		c = g.sysCli
 	}
 	// If for this client there is a recent matching subscription interest
 	// then we will map.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 4ff700ed..eb39fdbc 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -131,10 +131,14 @@ const (
 	// jsDirectGetPre
 	jsDirectGetPre = "$JS.API.DIRECT.GET"
 
-	// JSApiConsumerCreate is the endpoint to create ephemeral consumers for streams.
+	// JSApiConsumerCreate is the endpoint to create consumers for streams.
+	// This was also the legacy endpoint for ephemeral consumers.
+	// It now can take consumer name and optional filter subject, which when part of the subject controls access.
 	// Will return JSON response.
-	JSApiConsumerCreate  = "$JS.API.CONSUMER.CREATE.*"
-	JSApiConsumerCreateT = "$JS.API.CONSUMER.CREATE.%s"
+	JSApiConsumerCreate    = "$JS.API.CONSUMER.CREATE.*"
+	JSApiConsumerCreateT   = "$JS.API.CONSUMER.CREATE.%s"
+	JSApiConsumerCreateEx  = "$JS.API.CONSUMER.CREATE.*.>"
+	JSApiConsumerCreateExT = "$JS.API.CONSUMER.CREATE.%s.%s.%s"
 
 	// JSApiDurableCreate is the endpoint to create durable consumers for streams.
 	// You need to include the stream and consumer name in the subject.
@@ -832,8 +836,9 @@ func (s *Server) setJetStreamExportSubs() error {
 		{JSApiConsumerLeaderStepDown, s.jsConsumerLeaderStepDownRequest},
 		{JSApiMsgDelete, s.jsMsgDeleteRequest},
 		{JSApiMsgGet, s.jsMsgGetRequest},
+		{JSApiConsumerCreateEx, s.jsConsumerCreateRequest},
 		{JSApiConsumerCreate, s.jsConsumerCreateRequest},
-		{JSApiDurableCreate, s.jsDurableCreateRequest},
+		{JSApiDurableCreate, s.jsConsumerCreateRequest},
 		{JSApiConsumers, s.jsConsumerNamesRequest},
 		{JSApiConsumerList, s.jsConsumerListRequest},
 		{JSApiConsumerInfo, s.jsConsumerInfoRequest},
@@ -3609,17 +3614,19 @@ done:
 	mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, nil, nil, 0))
 }
 
-// Request to create a durable consumer.
-func (s *Server) jsDurableCreateRequest(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
-	s.jsConsumerCreate(sub, c, acc, subject, reply, msg, true)
-}
+// For determining consumer request type.
+type ccReqType uint8
 
-// Request to create a consumer.
-func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
-	s.jsConsumerCreate(sub, c, acc, subject, reply, msg, false)
-}
+const (
+	ccNew = iota
+	ccLegacyEphemeral
+	ccLegacyDurable
+)
 
-func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subject, reply string, rmsg []byte, expectDurable bool) {
+// Request to create a consumer where stream and optional consumer name are part of the subject, and optional
+// filtered subjects can be at the tail end.
+// Assumes stream and consumer names are single tokens.
+func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -3632,13 +3639,6 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 
 	var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
 
-	var streamName string
-	if expectDurable {
-		streamName = tokenAt(subject, 6)
-	} else {
-		streamName = tokenAt(subject, 5)
-	}
-
 	var req CreateConsumerRequest
 	if err := json.Unmarshal(msg, &req); err != nil {
 		resp.Error = NewJSInvalidJSONError()
@@ -3653,7 +3653,7 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 	if isClustered {
 		if req.Config.Direct {
 			// Check to see if we have this stream and are the stream leader.
-			if !acc.JetStreamIsStreamLeader(streamName) {
+			if !acc.JetStreamIsStreamLeader(streamNameFromSubject(subject)) {
 				return
 			}
 		} else {
@@ -3674,6 +3674,38 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 		}
 	}
 
+	var streamName, consumerName, filteredSubject string
+	var rt ccReqType
+
+	if n := numTokens(subject); n < 5 {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	} else if n == 5 {
+		// Legacy ephemeral.
+		rt = ccLegacyEphemeral
+		streamName = streamNameFromSubject(subject)
+	} else {
+		// New style and durable legacy.
+		if tokenAt(subject, 4) == "DURABLE" {
+			rt = ccLegacyDurable
+			if n != 7 {
+				resp.Error = NewJSConsumerDurableNameNotInSubjectError()
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+				return
+			}
+			streamName = tokenAt(subject, 6)
+			consumerName = tokenAt(subject, 7)
+		} else {
+			streamName = streamNameFromSubject(subject)
+			consumerName = consumerNameFromSubject(subject)
+		}
+		// New has optional filtered subject as part of main subject..
+		if n > 7 {
+			tokens := strings.Split(subject, tsep)
+			filteredSubject = strings.Join(tokens[6:], tsep)
+		}
+	}
+
 	if hasJS, doErr := acc.checkJetStream(); !hasJS {
 		if doErr {
 			resp.Error = NewJSNotEnabledForAccountError()
@@ -3681,14 +3713,25 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 		}
 		return
 	}
+
 	if streamName != req.Stream {
 		resp.Error = NewJSStreamMismatchError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
 
-	if expectDurable {
-		if numTokens(subject) != 7 {
+	if consumerName != _EMPTY_ {
+		// Check for path like separators in the name.
+		if strings.ContainsAny(consumerName, `\/`) {
+			resp.Error = NewJSConsumerNameContainsPathSeparatorsError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+	}
+
+	// Should we expect a durable name
+	if rt == ccLegacyDurable {
+		if numTokens(subject) < 7 {
 			resp.Error = NewJSConsumerDurableNameNotInSubjectError()
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
@@ -3699,30 +3742,36 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		}
-		consumerName := tokenAt(subject, 7)
 		if consumerName != req.Config.Durable {
 			resp.Error = NewJSConsumerDurableNameNotMatchSubjectError()
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		}
-		// Check for path like separators in the name.
-		if strings.ContainsAny(consumerName, `\/`) {
-			resp.Error = NewJSConsumerNameContainsPathSeparatorsError()
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-
-	} else {
-		if numTokens(subject) != 5 {
-			resp.Error = NewJSConsumerEphemeralWithDurableInSubjectError()
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
+	}
+	// If new style and durable set make sure they match.
+	if rt == ccNew {
 		if req.Config.Durable != _EMPTY_ {
-			resp.Error = NewJSConsumerEphemeralWithDurableNameError()
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
+			if consumerName != req.Config.Durable {
+				resp.Error = NewJSConsumerDurableNameNotMatchSubjectError()
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+				return
+			}
 		}
+		// New style ephemeral so we need to honor the name.
+		req.Config.Name = consumerName
+	}
+	// Check for legacy ephemeral mis-configuration.
+	if rt == ccLegacyEphemeral && req.Config.Durable != _EMPTY_ {
+		resp.Error = NewJSConsumerEphemeralWithDurableNameError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	// Check for a filter subject.
+	if filteredSubject != _EMPTY_ && req.Config.FilterSubject != filteredSubject {
+		resp.Error = NewJSConsumerCreateFilterSubjectMismatchError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
 	}
 
 	if isClustered && !req.Config.Direct {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index dfa7976f..e10f842c 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -851,12 +851,10 @@ func (js *jetStream) isMetaRecovering() bool {
 	return js.metaRecovering
 }
 
-// During recovery track any stream and consumer delete and update operations.
-type recoveryUpdates struct {
-	removeStreams   map[string]*streamAssignment
-	removeConsumers map[string]*consumerAssignment
-	updateStreams   map[string]*streamAssignment
-	updateConsumers map[string]*consumerAssignment
+// During recovery track any stream and consumer delete operations.
+type recoveryRemovals struct {
+	streams   map[string]*streamAssignment
+	consumers map[string]*consumerAssignment
 }
 
 func (js *jetStream) monitorCluster() {
@@ -901,11 +899,9 @@ func (js *jetStream) monitorCluster() {
 		}
 	}
 
-	ru := &recoveryUpdates{
-		removeStreams:   make(map[string]*streamAssignment),
-		removeConsumers: make(map[string]*consumerAssignment),
-		updateStreams:   make(map[string]*streamAssignment),
-		updateConsumers: make(map[string]*consumerAssignment),
+	rm := &recoveryRemovals{
+		streams:   make(map[string]*streamAssignment),
+		consumers: make(map[string]*consumerAssignment),
 	}
 
 	for {
@@ -922,27 +918,20 @@ func (js *jetStream) monitorCluster() {
 					js.clearMetaRecovering()
 
 					// Process any removes that are still valid after recovery.
-					for _, ca := range ru.removeConsumers {
+					for _, ca := range rm.consumers {
 						js.processConsumerRemoval(ca)
 					}
-					for _, sa := range ru.removeStreams {
+					for _, sa := range rm.streams {
 						js.processStreamRemoval(sa)
 					}
-					// Same with any pending updates.
-					for _, sa := range ru.updateStreams {
-						js.processUpdateStreamAssignment(sa)
-					}
-					for _, ca := range ru.updateConsumers {
-						js.processConsumerAssignment(ca)
-					}
 					// Clear.
-					ru = nil
+					rm = nil
 					s.Debugf("Recovered JetStream cluster metadata")
 					continue
 				}
 				ce := cei.(*CommittedEntry)
 				// FIXME(dlc) - Deal with errors.
-				if didSnap, didRemoval, err := js.applyMetaEntries(ce.Entries, ru); err == nil {
+				if didSnap, didRemoval, err := js.applyMetaEntries(ce.Entries, rm); err == nil {
 					_, nb := n.Applied(ce.Index)
 					if js.hasPeerEntries(ce.Entries) || didSnap || (didRemoval && time.Since(lastSnapTime) > 2*time.Second) {
 						// Since we received one make sure we have our own since we do not store
@@ -1384,23 +1373,7 @@ func (js *jetStream) hasPeerEntries(entries []*Entry) bool {
 	return false
 }
 
-const ksep = ":"
-
-func (sa *streamAssignment) recoveryKey() string {
-	if sa == nil {
-		return _EMPTY_
-	}
-	return sa.Client.serviceAccount() + ksep + sa.Config.Name
-}
-
-func (ca *consumerAssignment) recoveryKey() string {
-	if ca == nil {
-		return _EMPTY_
-	}
-	return ca.Client.serviceAccount() + ksep + ca.Stream + ksep + ca.Name
-}
-
-func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bool, bool, error) {
+func (js *jetStream) applyMetaEntries(entries []*Entry, rm *recoveryRemovals) (bool, bool, error) {
 	var didSnap, didRemove bool
 	isRecovering := js.isMetaRecovering()
 
@@ -1427,7 +1400,8 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				}
 				if isRecovering {
 					js.setStreamAssignmentRecovering(sa)
-					delete(ru.removeStreams, sa.recoveryKey())
+					key := sa.Client.Account + ":" + sa.Config.Name
+					delete(rm.streams, key)
 				}
 				if js.processStreamAssignment(sa) {
 					didRemove = true
@@ -1440,9 +1414,8 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				}
 				if isRecovering {
 					js.setStreamAssignmentRecovering(sa)
-					key := sa.recoveryKey()
-					ru.removeStreams[key] = sa
-					delete(ru.updateStreams, key)
+					key := sa.Client.Account + ":" + sa.Config.Name
+					rm.streams[key] = sa
 				} else {
 					js.processStreamRemoval(sa)
 					didRemove = true
@@ -1455,12 +1428,10 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				}
 				if isRecovering {
 					js.setConsumerAssignmentRecovering(ca)
-					key := ca.recoveryKey()
-					delete(ru.removeConsumers, key)
-					ru.updateConsumers[key] = ca
-				} else {
-					js.processConsumerAssignment(ca)
+					key := ca.Client.Account + ":" + ca.Name
+					delete(rm.consumers, key)
 				}
+				js.processConsumerAssignment(ca)
 			case assignCompressedConsumerOp:
 				ca, err := decodeConsumerAssignmentCompressed(buf[1:])
 				if err != nil {
@@ -1469,12 +1440,10 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				}
 				if isRecovering {
 					js.setConsumerAssignmentRecovering(ca)
-					key := ca.recoveryKey()
-					delete(ru.removeConsumers, key)
-					ru.updateConsumers[key] = ca
-				} else {
-					js.processConsumerAssignment(ca)
+					key := ca.Client.Account + ":" + ca.Name
+					delete(rm.consumers, key)
 				}
+				js.processConsumerAssignment(ca)
 			case removeConsumerOp:
 				ca, err := decodeConsumerAssignment(buf[1:])
 				if err != nil {
@@ -1483,9 +1452,8 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				}
 				if isRecovering {
 					js.setConsumerAssignmentRecovering(ca)
-					key := ca.recoveryKey()
-					ru.removeConsumers[key] = ca
-					delete(ru.updateConsumers, key)
+					key := ca.Client.Account + ":" + ca.Name
+					rm.consumers[key] = ca
 				} else {
 					js.processConsumerRemoval(ca)
 					didRemove = true
@@ -1498,12 +1466,9 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				}
 				if isRecovering {
 					js.setStreamAssignmentRecovering(sa)
-					key := sa.recoveryKey()
-					ru.updateStreams[key] = sa
-					delete(ru.removeStreams, key)
-				} else {
-					js.processUpdateStreamAssignment(sa)
 				}
+
+				js.processUpdateStreamAssignment(sa)
 			default:
 				panic("JetStream Cluster Unknown meta entry op type")
 			}
@@ -6339,6 +6304,7 @@ func (mset *stream) stateSnapshotLocked() []byte {
 		Deleted:  state.Deleted,
 	}
 	b, _ := json.Marshal(snap)
+
 	return b
 }
 
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 91419f93..cd8cb5eb 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -53,6 +53,9 @@ const (
 	// JSConsumerCreateErrF General consumer creation failure string ({err})
 	JSConsumerCreateErrF ErrorIdentifier = 10012
 
+	// JSConsumerCreateFilterSubjectMismatchErr Consumer create request did not match filtered subject from create subject
+	JSConsumerCreateFilterSubjectMismatchErr ErrorIdentifier = 10131
+
 	// JSConsumerDeliverCycleErr consumer deliver subject forms a cycle
 	JSConsumerDeliverCycleErr ErrorIdentifier = 10081
 
@@ -411,6 +414,7 @@ var (
 		JSConsumerBadDurableNameErr:                {Code: 400, ErrCode: 10103, Description: "durable name can not contain '.', '*', '>'"},
 		JSConsumerConfigRequiredErr:                {Code: 400, ErrCode: 10078, Description: "consumer config required"},
 		JSConsumerCreateErrF:                       {Code: 500, ErrCode: 10012, Description: "{err}"},
+		JSConsumerCreateFilterSubjectMismatchErr:   {Code: 400, ErrCode: 10131, Description: "Consumer create request did not match filtered subject from create subject"},
 		JSConsumerDeliverCycleErr:                  {Code: 400, ErrCode: 10081, Description: "consumer deliver subject forms a cycle"},
 		JSConsumerDeliverToWildcardsErr:            {Code: 400, ErrCode: 10079, Description: "consumer deliver subject has wildcards"},
 		JSConsumerDescriptionTooLongErrF:           {Code: 400, ErrCode: 10107, Description: "consumer description is too long, maximum allowed is {max}"},
@@ -721,6 +725,16 @@ func NewJSConsumerCreateError(err error, opts ...ErrorOption) *ApiError {
 	}
 }
 
+// NewJSConsumerCreateFilterSubjectMismatchError creates a new JSConsumerCreateFilterSubjectMismatchErr error: "Consumer create request did not match filtered subject from create subject"
+func NewJSConsumerCreateFilterSubjectMismatchError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSConsumerCreateFilterSubjectMismatchErr]
+}
+
 // NewJSConsumerDeliverCycleError creates a new JSConsumerDeliverCycleErr error: "consumer deliver subject forms a cycle"
 func NewJSConsumerDeliverCycleError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 846b7b63..732b7cea 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1487,6 +1487,8 @@ func TestFileStoreMeta(t *testing.T) {
 	if err := json.Unmarshal(buf, &oconfig2); err != nil {
 		t.Fatalf("Error unmarshalling: %v", err)
 	}
+	// Since we set name we will get that back now.
+	oconfig.Name = oname
 	if !reflect.DeepEqual(oconfig2, oconfig) {
 		t.Fatalf("Consumer configs not equal, got %+v vs %+v", oconfig2, oconfig)
 	}
@@ -3506,9 +3508,7 @@ func TestFileStorePurgeExKeepOneBug(t *testing.T) {
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, BlockSize: 128},
-		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 128}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -4512,102 +4512,3 @@ func TestFileStoreNoFSSAfterRecover(t *testing.T) {
 		t.Fatalf("Expected no fss post recover")
 	}
 }
-
-func TestFileStoreFSSCloseAndKeepOnExpireOnRecoverBug(t *testing.T) {
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer os.RemoveAll(storeDir)
-
-	ttl := 100 * time.Millisecond
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage, MaxAge: ttl},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	_, _, err = fs.StoreMsg("foo", nil, nil)
-	require_NoError(t, err)
-
-	fs.Stop()
-
-	time.Sleep(2 * ttl)
-
-	fs, err = newFileStore(
-		FileStoreConfig{StoreDir: storeDir},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage, MaxAge: ttl},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	if state := fs.State(); state.NumSubjects != 0 {
-		t.Fatalf("Expected no subjects with no messages, got %d", state.NumSubjects)
-	}
-}
-
-func TestFileStoreFSSBadStateBug(t *testing.T) {
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer os.RemoveAll(storeDir)
-
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	_, _, err = fs.StoreMsg("foo", nil, nil)
-	require_NoError(t, err)
-	_, _, err = fs.StoreMsg("foo", nil, nil)
-	require_NoError(t, err)
-
-	// Force write of fss.
-	mb := fs.getFirstBlock()
-	mb.mu.Lock()
-	mb.writePerSubjectInfo()
-	fssFile := filepath.Join(storeDir, msgDir, fmt.Sprintf(fssScan, 1))
-	buf, err := os.ReadFile(fssFile)
-	require_NoError(t, err)
-	mb.mu.Unlock()
-
-	// Now remove one of them.
-	fs.RemoveMsg(1)
-	fs.Stop()
-
-	// Now put back wrong fss with msgs == 2
-	err = os.WriteFile(fssFile, buf, defaultFilePerms)
-	require_NoError(t, err)
-
-	fs, err = newFileStore(
-		FileStoreConfig{StoreDir: storeDir},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	if fss := fs.SubjectsState("foo")["foo"]; fss.Msgs != 1 {
-		t.Fatalf("Got bad state on restart: %+v", fss)
-	}
-}
-
-func TestFileStoreFSSExpireNumPendingBug(t *testing.T) {
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer os.RemoveAll(storeDir)
-
-	cexp := 100 * time.Millisecond
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp},
-		StreamConfig{Name: "zzz", Subjects: []string{"KV.>"}, MaxMsgsPer: 1, Storage: FileStorage},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	// Let FSS meta expire.
-	time.Sleep(2 * cexp)
-
-	_, _, err = fs.StoreMsg("KV.X", nil, []byte("Y"))
-	require_NoError(t, err)
-
-	if fss := fs.FilteredState(1, "KV.X"); fss.Msgs != 1 {
-		t.Fatalf("Expected only 1 msg, got %d", fss.Msgs)
-	}
-}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index c873f8be..be203885 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -12982,87 +12982,3 @@ func TestJetStreamClusterStaleReadsOnRestart(t *testing.T) {
 		}
 	}
 }
-
-func TestJetStreamClusterReplicasChangeStreamInfo(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3F", 3)
-	defer c.shutdown()
-
-	s := c.randomServer()
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	numStreams := 1
-	msgsPerStream := 10
-	for i := 0; i < numStreams; i++ {
-		sname := fmt.Sprintf("TEST_%v", i)
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:     sname,
-			Replicas: 3,
-		})
-		require_NoError(t, err)
-		for j := 0; j < msgsPerStream; j++ {
-			sendStreamMsg(t, nc, sname, "msg")
-		}
-	}
-
-	checkStreamInfo := func(js nats.JetStreamContext) {
-		t.Helper()
-		checkFor(t, 20*time.Second, 15*time.Millisecond, func() error {
-			for i := 0; i < numStreams; i++ {
-				si, err := js.StreamInfo(fmt.Sprintf("TEST_%v", i))
-				if err != nil {
-					return err
-				}
-				if si.State.Msgs != uint64(msgsPerStream) || si.State.FirstSeq != 1 || si.State.LastSeq != uint64(msgsPerStream) {
-					return fmt.Errorf("Invalid stream info for %s: %+v", si.Config.Name, si.State)
-				}
-			}
-			return nil
-		})
-	}
-	checkStreamInfo(js)
-
-	// Update replicas down to 1
-	for i := 0; i < numStreams; i++ {
-		sname := fmt.Sprintf("TEST_%v", i)
-		_, err := js.UpdateStream(&nats.StreamConfig{
-			Name:     sname,
-			Replicas: 1,
-		})
-		require_NoError(t, err)
-	}
-	checkStreamInfo(js)
-
-	// Back up to 3
-	for i := 0; i < numStreams; i++ {
-		sname := fmt.Sprintf("TEST_%v", i)
-		_, err := js.UpdateStream(&nats.StreamConfig{
-			Name:     sname,
-			Replicas: 3,
-		})
-		require_NoError(t, err)
-		c.waitOnStreamLeader(globalAccountName, sname)
-		for _, s := range c.servers {
-			c.waitOnStreamCurrent(s, globalAccountName, sname)
-		}
-	}
-	checkStreamInfo(js)
-
-	// Now shutdown the cluster and restart it
-	nc.Close()
-	c.stopAll()
-	c.restartAll()
-
-	for i := 0; i < numStreams; i++ {
-		sname := fmt.Sprintf("TEST_%v", i)
-		c.waitOnStreamLeader(globalAccountName, sname)
-		for _, s := range c.servers {
-			c.waitOnStreamCurrent(s, globalAccountName, sname)
-		}
-	}
-
-	nc, js = jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	checkStreamInfo(js)
-}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index aa0d043f..9c92c4b6 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -19284,3 +19284,129 @@ func TestJetStreamConsumerEOFBugNewFileStore(t *testing.T) {
 	_, err = sub.NextMsg(time.Second)
 	require_NoError(t, err)
 }
+
+func TestJetStreamSubjectBasedFilteredConsumers(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		jetstream: {max_mem_store: 64GB, max_file_store: 10TB}
+		accounts: {
+			A: {
+				jetstream: enabled
+				users: [ {
+					user: u,
+					password: p
+					permissions {
+						publish {
+							allow: [
+								'ID.>',
+								'$JS.API.INFO',
+								'$JS.API.STREAM.>',
+								'$JS.API.CONSUMER.INFO.>',
+								'$JS.API.CONSUMER.CREATE.TEST.VIN-xxx.ID.foo.>', # Only allow ID.foo.
+							]
+							deny: [ '$JS.API.CONSUMER.CREATE.*', '$JS.API.CONSUMER.DURABLE.CREATE.*.*']
+						}
+					}
+				} ]
+			},
+		}
+	`))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s, nats.UserInfo("u", "p"), nats.ErrorHandler(noOpErrHandler))
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"ID.*.*"},
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 100; i++ {
+		js.Publish(fmt.Sprintf("ID.foo.%d", i*3), nil)
+		js.Publish(fmt.Sprintf("ID.bar.%d", i*3+1), nil)
+		js.Publish(fmt.Sprintf("ID.baz.%d", i*3+2), nil)
+	}
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_True(t, si.State.Msgs == 300)
+
+	// Trying to create a consumer with non filtered API should fail.
+	js, err = nc.JetStream(nats.MaxWait(200 * time.Millisecond))
+	require_NoError(t, err)
+
+	_, err = js.SubscribeSync("ID.foo.*")
+	require_Error(t, err, nats.ErrTimeout)
+
+	_, err = js.SubscribeSync("ID.foo.*", nats.Durable("dlc"))
+	require_Error(t, err, nats.ErrTimeout)
+
+	// Direct filtered should work.
+	// Need to do by hand for now.
+	ecSubj := fmt.Sprintf(JSApiConsumerCreateExT, "TEST", "VIN-xxx", "ID.foo.*")
+
+	crReq := CreateConsumerRequest{
+		Stream: "TEST",
+		Config: ConsumerConfig{
+			DeliverPolicy: DeliverLast,
+			FilterSubject: "ID.foo.*",
+			AckPolicy:     AckExplicit,
+		},
+	}
+	req, err := json.Marshal(crReq)
+	require_NoError(t, err)
+
+	resp, err := nc.Request(ecSubj, req, 500*time.Millisecond)
+	require_NoError(t, err)
+	var ccResp JSApiConsumerCreateResponse
+	err = json.Unmarshal(resp.Data, &ccResp)
+	require_NoError(t, err)
+	if ccResp.Error != nil {
+		t.Fatalf("Unexpected error: %v", ccResp.Error)
+	}
+	cfg := ccResp.Config
+	ci := ccResp.ConsumerInfo
+	// Make sure we recognized as an ephemeral (since no durable was set) and that we have an InactiveThreshold.
+	// Make sure we captured preferred ephemeral name.
+	if ci.Name != "VIN-xxx" {
+		t.Fatalf("Did not get correct name, expected %q got %q", "xxx", ci.Name)
+	}
+	if cfg.InactiveThreshold == 0 {
+		t.Fatalf("Expected default inactive threshold to be set, got %v", cfg.InactiveThreshold)
+	}
+
+	// Make sure we can not use different consumer name since locked above.
+	ecSubj = fmt.Sprintf(JSApiConsumerCreateExT, "TEST", "VIN-zzz", "ID.foo.*")
+	_, err = nc.Request(ecSubj, req, 500*time.Millisecond)
+	require_Error(t, err, nats.ErrTimeout)
+
+	// Now check that we error when we mismatch filtersubject.
+	crReq = CreateConsumerRequest{
+		Stream: "TEST",
+		Config: ConsumerConfig{
+			DeliverPolicy: DeliverLast,
+			FilterSubject: "ID.bar.*",
+			AckPolicy:     AckExplicit,
+		},
+	}
+	req, err = json.Marshal(crReq)
+	require_NoError(t, err)
+
+	ecSubj = fmt.Sprintf(JSApiConsumerCreateExT, "TEST", "VIN-xxx", "ID.foo.*")
+	resp, err = nc.Request(ecSubj, req, 500*time.Millisecond)
+	require_NoError(t, err)
+	err = json.Unmarshal(resp.Data, &ccResp)
+	require_NoError(t, err)
+	checkNatsError(t, ccResp.Error, JSConsumerCreateFilterSubjectMismatchErr)
+
+	// Now make sure if we change subject to match that we can not create a filtered consumer on ID.bar.>
+	ecSubj = fmt.Sprintf(JSApiConsumerCreateExT, "TEST", "VIN-xxx", "ID.bar.*")
+	_, err = nc.Request(ecSubj, req, 500*time.Millisecond)
+	require_Error(t, err, nats.ErrTimeout)
+}
diff --git a/server/norace_test.go b/server/norace_test.go
index 1b3d7a03..ff730b4d 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -5399,7 +5399,7 @@ func TestNoRaceJetStreamFileStoreLargeKVAccessTiming(t *testing.T) {
 
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize, CacheExpire: 5 * time.Second},
-		StreamConfig{Name: "zzz", Subjects: []string{"KV.STREAM_NAME.*"}, Storage: FileStorage, MaxMsgsPer: 1},
+		StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgsPer: 1},
 	)
 	require_NoError(t, err)
 	defer fs.Stop()