Repository: nats-io/nats-server
Pull request #2373 (2021-07-25 13:44:15)

Remove the swapping of accounts during processing of service imports.
When processing service imports we would swap out the accounts during processing.
With the addition of internal subscriptions and internal clients publishing in JetStream we had an issue with the wrong account being used.

This was specific to delayed pull subscribers trying to unsubscribe due to max of 1 while other JetStream API calls were running concurrently.

/cc @nats-io/core



Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 2a2bce20..9fe4a81c 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1869,8 +1869,8 @@ func (a *Account) addServiceImportSub(si *serviceImport) error {
 	subject := si.from
 	a.mu.Unlock()
 
-	cb := func(sub *subscription, c *client, subject, reply string, msg []byte) {
-		c.processServiceImport(si, a, msg)
+	cb := func(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
+		c.processServiceImport(si, acc, msg)
 	}
 	sub, err := c.processSubEx([]byte(subject), nil, []byte(sid), cb, true, true, false)
 	if err != nil {
@@ -2038,7 +2038,7 @@ const (
 )
 
 // This is where all service export responses are handled.
-func (a *Account) processServiceImportResponse(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (a *Account) processServiceImportResponse(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	a.mu.RLock()
 	if a.expired || len(a.exports.responses) == 0 {
 		a.mu.RUnlock()
@@ -3679,7 +3679,7 @@ func (dr *DirAccResolver) Start(s *Server) error {
 	packRespIb := s.newRespInbox()
 	for _, reqSub := range []string{accUpdateEventSubjOld, accUpdateEventSubjNew} {
 		// subscribe to account jwt update requests
-		if _, err := s.sysSubscribe(fmt.Sprintf(reqSub, "*"), func(_ *subscription, _ *client, subj, resp string, msg []byte) {
+		if _, err := s.sysSubscribe(fmt.Sprintf(reqSub, "*"), func(_ *subscription, _ *client, _ *Account, subj, resp string, msg []byte) {
 			pubKey := ""
 			tk := strings.Split(subj, tsep)
 			if len(tk) == accUpdateTokensNew {
@@ -3709,7 +3709,7 @@ func (dr *DirAccResolver) Start(s *Server) error {
 			return fmt.Errorf("error setting up update handling: %v", err)
 		}
 	}
-	if _, err := s.sysSubscribe(accClaimsReqSubj, func(_ *subscription, _ *client, subj, resp string, msg []byte) {
+	if _, err := s.sysSubscribe(accClaimsReqSubj, func(_ *subscription, _ *client, _ *Account, subj, resp string, msg []byte) {
 		if claim, err := jwt.DecodeAccountClaims(string(msg)); err != nil {
 			respondToUpdate(s, resp, "n/a", "jwt update resulted in error", err)
 		} else if claim.Issuer == op && strict {
@@ -3726,7 +3726,7 @@ func (dr *DirAccResolver) Start(s *Server) error {
 		return fmt.Errorf("error setting up update handling: %v", err)
 	}
 	// respond to lookups with our version
-	if _, err := s.sysSubscribe(fmt.Sprintf(accLookupReqSubj, "*"), func(_ *subscription, _ *client, subj, reply string, msg []byte) {
+	if _, err := s.sysSubscribe(fmt.Sprintf(accLookupReqSubj, "*"), func(_ *subscription, _ *client, _ *Account, subj, reply string, msg []byte) {
 		if reply == "" {
 			return
 		}
@@ -3744,7 +3744,7 @@ func (dr *DirAccResolver) Start(s *Server) error {
 	}
 	// respond to pack requests with one or more pack messages
 	// an empty message signifies the end of the response responder
-	if _, err := s.sysSubscribeQ(accPackReqSubj, "responder", func(_ *subscription, _ *client, _, reply string, theirHash []byte) {
+	if _, err := s.sysSubscribeQ(accPackReqSubj, "responder", func(_ *subscription, _ *client, _ *Account, _, reply string, theirHash []byte) {
 		if reply == "" {
 			return
 		}
@@ -3765,18 +3765,18 @@ func (dr *DirAccResolver) Start(s *Server) error {
 		return fmt.Errorf("error setting up pack request handling: %v", err)
 	}
 	// respond to list requests with one message containing all account ids
-	if _, err := s.sysSubscribe(accListReqSubj, func(_ *subscription, _ *client, _, reply string, _ []byte) {
+	if _, err := s.sysSubscribe(accListReqSubj, func(_ *subscription, _ *client, _ *Account, _, reply string, _ []byte) {
 		handleListRequest(dr.DirJWTStore, s, reply)
 	}); err != nil {
 		return fmt.Errorf("error setting up list request handling: %v", err)
 	}
-	if _, err := s.sysSubscribe(accDeleteReqSubj, func(_ *subscription, _ *client, _, reply string, msg []byte) {
+	if _, err := s.sysSubscribe(accDeleteReqSubj, func(_ *subscription, _ *client, _ *Account, _, reply string, msg []byte) {
 		handleDeleteRequest(dr.DirJWTStore, s, msg, reply)
 	}); err != nil {
 		return fmt.Errorf("error setting up delete request handling: %v", err)
 	}
 	// embed pack responses into store
-	if _, err := s.sysSubscribe(packRespIb, func(_ *subscription, _ *client, _, _ string, msg []byte) {
+	if _, err := s.sysSubscribe(packRespIb, func(_ *subscription, _ *client, _ *Account, _, _ string, msg []byte) {
 		hash := dr.DirJWTStore.Hash()
 		if len(msg) == 0 { // end of response stream
 			s.Debugf("Merging Finished and resulting in: %x", dr.DirJWTStore.Hash())
@@ -3894,7 +3894,7 @@ func (s *Server) fetch(res AccountResolver, name string, timeout time.Duration)
 	replySubj := s.newRespInbox()
 	replies := s.sys.replies
 	// Store our handler.
-	replies[replySubj] = func(sub *subscription, _ *client, subject, _ string, msg []byte) {
+	replies[replySubj] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 		clone := make([]byte, len(msg))
 		copy(clone, msg)
 		s.mu.Lock()
@@ -3965,7 +3965,7 @@ func (dr *CacheDirAccResolver) Start(s *Server) error {
 	}
 	for _, reqSub := range []string{accUpdateEventSubjOld, accUpdateEventSubjNew} {
 		// subscribe to account jwt update requests
-		if _, err := s.sysSubscribe(fmt.Sprintf(reqSub, "*"), func(_ *subscription, _ *client, subj, resp string, msg []byte) {
+		if _, err := s.sysSubscribe(fmt.Sprintf(reqSub, "*"), func(_ *subscription, _ *client, _ *Account, subj, resp string, msg []byte) {
 			pubKey := ""
 			tk := strings.Split(subj, tsep)
 			if len(tk) == accUpdateTokensNew {
@@ -3997,7 +3997,7 @@ func (dr *CacheDirAccResolver) Start(s *Server) error {
 			return fmt.Errorf("error setting up update handling: %v", err)
 		}
 	}
-	if _, err := s.sysSubscribe(accClaimsReqSubj, func(_ *subscription, _ *client, subj, resp string, msg []byte) {
+	if _, err := s.sysSubscribe(accClaimsReqSubj, func(_ *subscription, _ *client, _ *Account, subj, resp string, msg []byte) {
 		if claim, err := jwt.DecodeAccountClaims(string(msg)); err != nil {
 			respondToUpdate(s, resp, "n/a", "jwt update cache resulted in error", err)
 		} else if claim.Issuer == op && strict {
@@ -4016,12 +4016,12 @@ func (dr *CacheDirAccResolver) Start(s *Server) error {
 		return fmt.Errorf("error setting up update handling: %v", err)
 	}
 	// respond to list requests with one message containing all account ids
-	if _, err := s.sysSubscribe(accListReqSubj, func(_ *subscription, _ *client, _, reply string, _ []byte) {
+	if _, err := s.sysSubscribe(accListReqSubj, func(_ *subscription, _ *client, _ *Account, _, reply string, _ []byte) {
 		handleListRequest(dr.DirJWTStore, s, reply)
 	}); err != nil {
 		return fmt.Errorf("error setting up list request handling: %v", err)
 	}
-	if _, err := s.sysSubscribe(accDeleteReqSubj, func(_ *subscription, _ *client, _, reply string, msg []byte) {
+	if _, err := s.sysSubscribe(accDeleteReqSubj, func(_ *subscription, _ *client, _ *Account, _, reply string, msg []byte) {
 		handleDeleteRequest(dr.DirJWTStore, s, msg, reply)
 	}); err != nil {
 		return fmt.Errorf("error setting up list request handling: %v", err)
diff --git a/server/client.go b/server/client.go
index f1b60db3..25650e3a 100644
--- a/server/client.go
+++ b/server/client.go
@@ -422,7 +422,7 @@ func (c *client) String() (id string) {
 		return loaded.(string)
 	}
 
-	return ""
+	return _EMPTY_
 }
 
 // GetName returns the application supplied name for the connection.
@@ -733,7 +733,7 @@ func (c *client) applyAccountLimits() {
 	}
 	c.mpay = jwt.NoLimit
 	c.msubs = jwt.NoLimit
-	if c.opts.JWT != "" { // user jwt implies account
+	if c.opts.JWT != _EMPTY_ { // user jwt implies account
 		if uc, _ := jwt.DecodeUserClaims(c.opts.JWT); uc != nil {
 			c.mpay = int32(uc.Limits.Payload)
 			c.msubs = int32(uc.Limits.Subs)
@@ -1729,7 +1729,7 @@ func (c *client) processConnect(arg []byte) error {
 			}
 		}
 		if ncs != _EMPTY_ {
-			c.ncs.Store(fmt.Sprintf("%s - %q", c.String(), ncs))
+			c.ncs.Store(fmt.Sprintf("%s - %q", c, ncs))
 		}
 	}
 
@@ -3028,7 +3028,7 @@ var needFlush = struct{}{}
 
 // deliverMsg will deliver a message to a matching subscription and its underlying client.
 // We process all connection/client types. mh is the part that will be protocol/client specific.
-func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, gwrply bool) bool {
+func (c *client) deliverMsg(sub *subscription, acc *Account, subject, reply, mh, msg []byte, gwrply bool) bool {
 	if sub.client == nil {
 		return false
 	}
@@ -3132,9 +3132,9 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 
 		// Internal account clients are for service imports and need the '\r\n'.
 		if client.kind == ACCOUNT {
-			sub.icb(sub, c, string(subject), string(reply), msg)
+			sub.icb(sub, c, acc, string(subject), string(reply), msg)
 		} else {
-			sub.icb(sub, c, string(subject), string(reply), msg[:msgSize])
+			sub.icb(sub, c, acc, string(subject), string(reply), msg[:msgSize])
 		}
 		return true
 	}
@@ -3833,6 +3833,8 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 				nrr = []byte(rsi.from)
 			}
 		} else {
+			// This only happens when we do a pull subscriber that trampolines through another account.
+			// Normally this code is not called.
 			nrr = c.pa.reply
 		}
 	} else if !isResponse && si.latency != nil && tracking {
@@ -3852,7 +3854,6 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// Pick correct "to" subject. If we matched on a wildcard use the literal publish subject.
 	to, subject := si.to, string(c.pa.subject)
 
-	hadPrevSi := c.pa.psi != nil
 	if si.tr != nil {
 		// FIXME(dlc) - This could be slow, may want to look at adding cache to bare transforms?
 		to, _ = si.tr.transformSubject(subject)
@@ -3860,9 +3861,8 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 		to = subject
 	}
 
-	// Copy our pubArg and account
+	// Copy our pubArg since this gets modified as we process the service import itself.
 	pacopy := c.pa
-	oacc := c.acc
 
 	// Now check to see if this account has mappings that could affect the service import.
 	// Can't use non-locked trick like in processInboundClientMsg, so just call into selectMappedSubject
@@ -3872,9 +3872,8 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 		to = nsubj
 	}
 
-	// Change this so that we detect recursion
-	// Remember prior.
-	share := si.share
+	// Set previous service import to detect chaining.
+	hadPrevSi, share := c.pa.psi != nil, si.share
 	if hadPrevSi {
 		share = c.pa.psi.share
 	}
@@ -3888,7 +3887,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 			var cis ClientInfo
 			if err := json.Unmarshal(getHeader(ClientInfoHdr, msg[:c.pa.hdr]), &cis); err == nil {
 				ci = &cis
-				ci.Service = c.acc.Name
+				ci.Service = acc.Name
 			}
 		} else if c.kind != LEAF || c.pa.hdr < 0 || len(getHeader(ClientInfoHdr, msg[:c.pa.hdr])) == 0 {
 			ci = c.getClientInfo(share)
@@ -3906,9 +3905,6 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 		c.pa.subject = []byte(to)
 	}
 	c.pa.reply = nrr
-	c.mu.Lock()
-	c.acc = si.acc
-	c.mu.Unlock()
 
 	// FIXME(dlc) - Do L1 cache trick like normal client?
 	rr := si.acc.sl.Match(to)
@@ -3942,12 +3938,9 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 		didDeliver, _ = c.processMsgResults(si.acc, rr, msg, c.pa.deliver, []byte(to), nrr, flags)
 	}
 
-	// Put what was there back now.
+	// Restore to original values.
 	c.in.rts = orts
 	c.pa = pacopy
-	c.mu.Lock()
-	c.acc = oacc
-	c.mu.Unlock()
 
 	// Determine if we should remove this service import. This is for response service imports.
 	// We will remove if we did not deliver, or if we are a response service import and we are
@@ -4119,7 +4112,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 
 		// Normal delivery
 		mh := c.msgHeader(dsubj, creply, sub)
-		didDeliver = c.deliverMsg(sub, dsubj, creply, mh, msg, rplyHasGWPrefix) || didDeliver
+		didDeliver = c.deliverMsg(sub, acc, dsubj, creply, mh, msg, rplyHasGWPrefix) || didDeliver
 	}
 
 	// Set these up to optionally filter based on the queue lists.
@@ -4243,7 +4236,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 			}
 
 			mh := c.msgHeader(dsubj, creply, sub)
-			if c.deliverMsg(sub, subject, creply, mh, msg, rplyHasGWPrefix) {
+			if c.deliverMsg(sub, acc, subject, creply, mh, msg, rplyHasGWPrefix) {
 				didDeliver = true
 				// Clear rsub
 				rsub = nil
@@ -4320,7 +4313,7 @@ sendToRoutesOrLeafs:
 		}
 
 		mh := c.msgHeaderForRouteOrLeaf(subject, reply, rt, acc)
-		didDeliver = c.deliverMsg(rt.sub, subject, reply, mh, dmsg, false) || didDeliver
+		didDeliver = c.deliverMsg(rt.sub, acc, subject, reply, mh, dmsg, false) || didDeliver
 
 		// If we set the header reset the origin pub args.
 		if hset {
diff --git a/server/consumer.go b/server/consumer.go
index 086e8eec..63347f80 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -777,7 +777,7 @@ func (o *consumer) setLeader(isLeader bool) {
 	}
 }
 
-func (o *consumer) handleClusterConsumerInfoRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (o *consumer) handleClusterConsumerInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	o.mu.RLock()
 	sysc := o.sysc
 	o.mu.RUnlock()
@@ -1096,7 +1096,7 @@ func (o *consumer) sendAckReply(subj string) {
 }
 
 // Process a message for the ack reply subject delivered with a message.
-func (o *consumer) processAck(_ *subscription, c *client, subject, reply string, rmsg []byte) {
+func (o *consumer) processAck(_ *subscription, c *client, acc *Account, subject, reply string, rmsg []byte) {
 	_, msg := c.msgParts(rmsg)
 	sseq, dseq, dc := ackReplyInfo(subject)
 
@@ -1115,7 +1115,7 @@ func (o *consumer) processAck(_ *subscription, c *client, subject, reply string,
 		// somewhere else.
 		phdr := c.pa.hdr
 		c.pa.hdr = -1
-		o.processNextMsgReq(nil, c, subject, reply, msg[len(AckNext):])
+		o.processNextMsgReq(nil, c, acc, subject, reply, msg[len(AckNext):])
 		c.pa.hdr = phdr
 		skipAckReply = true
 	case bytes.Equal(msg, AckNak):
@@ -1763,7 +1763,7 @@ func (wq *waitQueue) pop() *waitingRequest {
 // 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, _, reply string, msg []byte) {
+func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _, reply string, msg []byte) {
 	_, msg = c.msgParts(msg)
 
 	o.mu.Lock()
@@ -2263,7 +2263,7 @@ func (o *consumer) needFlowControl() bool {
 	return false
 }
 
-func (o *consumer) processFlowControl(_ *subscription, c *client, subj, _ string, _ []byte) {
+func (o *consumer) processFlowControl(_ *subscription, c *client, _ *Account, subj, _ string, _ []byte) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
diff --git a/server/events.go b/server/events.go
index a3742fe4..41da9cc4 100644
--- a/server/events.go
+++ b/server/events.go
@@ -723,35 +723,35 @@ func (s *Server) initEventTracking() {
 	}
 	monSrvc := map[string]msgHandler{
 		"STATSZ": s.statszReq,
-		"VARZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"VARZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &VarzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Varz(&optz.VarzOptions) })
 		},
-		"SUBSZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"SUBSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &SubszEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Subsz(&optz.SubszOptions) })
 		},
-		"CONNZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"CONNZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &ConnzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Connz(&optz.ConnzOptions) })
 		},
-		"ROUTEZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"ROUTEZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &RoutezEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Routez(&optz.RoutezOptions) })
 		},
-		"GATEWAYZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"GATEWAYZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &GatewayzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Gatewayz(&optz.GatewayzOptions) })
 		},
-		"LEAFZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"LEAFZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &LeafzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Leafz(&optz.LeafzOptions) })
 		},
-		"ACCOUNTZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"ACCOUNTZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &AccountzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Accountz(&optz.AccountzOptions) })
 		},
-		"JSZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"JSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &JszEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Jsz(&optz.JSzOptions) })
 		},
@@ -774,7 +774,7 @@ func (s *Server) initEventTracking() {
 		}
 	}
 	monAccSrvc := map[string]msgHandler{
-		"SUBSZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"SUBSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &SubszEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(subject); err != nil {
@@ -786,7 +786,7 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"CONNZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"CONNZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &ConnzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(subject); err != nil {
@@ -797,7 +797,7 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"LEAFZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"LEAFZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &LeafzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(subject); err != nil {
@@ -808,7 +808,7 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"JSZ": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"JSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &JszEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(subject); err != nil {
@@ -819,7 +819,7 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"INFO": func(sub *subscription, c *client, subject, reply string, msg []byte) {
+		"INFO": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &AccInfoEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(subject); err != nil {
@@ -868,11 +868,11 @@ func (s *Server) addSystemAccountExports(sacc *Account) {
 }
 
 // accountClaimUpdate will receive claim updates for accounts.
-func (s *Server) accountClaimUpdate(sub *subscription, _ *client, subject, resp string, msg []byte) {
+func (s *Server) accountClaimUpdate(sub *subscription, _ *client, _ *Account, subject, resp string, msg []byte) {
 	if !s.EventsEnabled() {
 		return
 	}
-	pubKey := ""
+	var pubKey string
 	toks := strings.Split(subject, tsep)
 	if len(toks) == accUpdateTokensNew {
 		pubKey = toks[accReqAccIndex]
@@ -918,7 +918,7 @@ func (s *Server) processRemoteServerShutdown(sid string) {
 }
 
 // remoteServerShutdownEvent is called when we get an event from another server shutting down.
-func (s *Server) remoteServerShutdown(sub *subscription, _ *client, subject, reply string, msg []byte) {
+func (s *Server) remoteServerShutdown(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if !s.eventsEnabled() {
@@ -951,7 +951,7 @@ func (s *Server) remoteServerShutdown(sub *subscription, _ *client, subject, rep
 }
 
 // remoteServerUpdate listens for statsz updates from other servers.
-func (s *Server) remoteServerUpdate(sub *subscription, _ *client, subject, reply string, msg []byte) {
+func (s *Server) remoteServerUpdate(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	var ssm ServerStatsMsg
 	if err := json.Unmarshal(msg, &ssm); err != nil {
 		s.Debugf("Received bad server info for remote server update")
@@ -1045,7 +1045,7 @@ func (s *Server) shutdownEventing() {
 }
 
 // Request for our local connection count.
-func (s *Server) connsRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) connsRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
@@ -1084,7 +1084,7 @@ func (s *Server) connsRequest(sub *subscription, c *client, subject, reply strin
 }
 
 // leafNodeConnected is an event we will receive when a leaf node for a given account connects.
-func (s *Server) leafNodeConnected(sub *subscription, _ *client, subject, reply string, msg []byte) {
+func (s *Server) leafNodeConnected(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	m := accNumConnsReq{}
 	if err := json.Unmarshal(msg, &m); err != nil {
 		s.sys.client.Errorf("Error unmarshalling account connections request message: %v", err)
@@ -1092,7 +1092,7 @@ func (s *Server) leafNodeConnected(sub *subscription, _ *client, subject, reply
 	}
 
 	s.mu.Lock()
-	na := m.Account == "" || !s.eventsEnabled() || !s.gateway.enabled
+	na := m.Account == _EMPTY_ || !s.eventsEnabled() || !s.gateway.enabled
 	s.mu.Unlock()
 
 	if na {
@@ -1205,7 +1205,7 @@ func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
 }
 
 // statszReq is a request for us to respond with current statsz.
-func (s *Server) statszReq(sub *subscription, _ *client, subject, reply string, msg []byte) {
+func (s *Server) statszReq(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	if !s.EventsEnabled() || reply == _EMPTY_ {
 		return
 	}
@@ -1264,7 +1264,7 @@ func (s *Server) zReq(c *client, reply string, rmsg []byte, fOpts *EventFilterOp
 }
 
 // remoteConnsUpdate gets called when we receive a remote update from another server.
-func (s *Server) remoteConnsUpdate(sub *subscription, _ *client, subject, reply string, msg []byte) {
+func (s *Server) remoteConnsUpdate(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
@@ -1577,10 +1577,10 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 	s.mu.Unlock()
 }
 
-// Internal message callback. If the msg is needed past the callback it is
-// required to be copied.
+// Internal message callback.
+// If the msg is needed past the callback it is required to be copied.
 // rmsg contains header and the message. use client.msgParts(rmsg) to split them apart
-type msgHandler func(sub *subscription, client *client, subject, reply string, rmsg []byte)
+type msgHandler func(sub *subscription, client *client, acc *Account, subject, reply string, rmsg []byte)
 
 // Create an internal subscription. sysSubscribeQ for queue groups.
 func (s *Server) sysSubscribe(subject string, cb msgHandler) (*subscription, error) {
@@ -1656,7 +1656,7 @@ func remoteLatencySubjectForResponse(subject []byte) string {
 }
 
 // remoteLatencyUpdate is used to track remote latency measurements for tracking on exported services.
-func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, subject, _ string, msg []byte) {
+func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
@@ -1719,7 +1719,7 @@ func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, subject, _ st
 
 // This is used for all inbox replies so that we do not send supercluster wide interest
 // updates for every request. Same trick used in modern NATS clients.
-func (s *Server) inboxReply(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (s *Server) inboxReply(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
 	s.mu.Lock()
 	if !s.eventsEnabled() || s.sys.replies == nil {
 		s.mu.Unlock()
@@ -1729,7 +1729,7 @@ func (s *Server) inboxReply(sub *subscription, c *client, subject, reply string,
 	s.mu.Unlock()
 
 	if ok && cb != nil {
-		cb(sub, c, subject, reply, msg)
+		cb(sub, c, acc, subject, reply, msg)
 	}
 }
 
@@ -1792,7 +1792,7 @@ func totalSubs(rr *SublistResult, qg []byte) (nsubs int32) {
 
 // Allows users of large systems to debug active subscribers for a given subject.
 // Payload should be the subject of interest.
-func (s *Server) debugSubscribers(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) debugSubscribers(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	// Even though this is an internal only subscription, meaning interest was not forwarded, we could
 	// get one here from a GW in optimistic mode. Ignore for now.
 	// FIXME(dlc) - Should we send no interest here back to the GW?
@@ -1859,7 +1859,7 @@ func (s *Server) debugSubscribers(sub *subscription, c *client, subject, reply s
 	// Create direct reply inbox that we multiplex under the WC replies.
 	replySubj := s.newRespInbox()
 	// Store our handler.
-	s.sys.replies[replySubj] = func(sub *subscription, _ *client, subject, _ string, msg []byte) {
+	s.sys.replies[replySubj] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 		if n, err := strconv.Atoi(string(msg)); err == nil {
 			atomic.AddInt32(&nsubs, int32(n))
 		}
@@ -1902,7 +1902,7 @@ func (s *Server) debugSubscribers(sub *subscription, c *client, subject, reply s
 
 // Request for our local subscription count. This will come from a remote origin server
 // that received the initial request.
-func (s *Server) nsubsRequest(sub *subscription, _ *client, subject, reply string, msg []byte) {
+func (s *Server) nsubsRequest(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
diff --git a/server/gateway.go b/server/gateway.go
index 98c104a5..f84633c9 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -2552,7 +2552,7 @@ func (c *client) sendMsgToGateways(acc *Account, msg, subject, reply []byte, qgr
 		sub.nm, sub.max = 0, 0
 		sub.client = gwc
 		sub.subject = subject
-		didDeliver = c.deliverMsg(sub, subject, mreply, mh, msg, false) || didDeliver
+		didDeliver = c.deliverMsg(sub, acc, subject, mreply, mh, msg, false) || didDeliver
 	}
 	// Done with subscription, put back to pool. We don't need
 	// to reset content since we explicitly set when using it.
diff --git a/server/jetstream.go b/server/jetstream.go
index e11fa696..0324b18d 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1355,7 +1355,7 @@ func (a *Account) JetStreamEnabled() bool {
 	return enabled
 }
 
-func (jsa *jsAccount) remoteUpdateUsage(sub *subscription, c *client, subject, _ string, msg []byte) {
+func (jsa *jsAccount) remoteUpdateUsage(sub *subscription, c *client, _ *Account, subject, _ string, msg []byte) {
 	const usageSize = 32
 
 	jsa.mu.Lock()
@@ -1826,7 +1826,7 @@ func (t *streamTemplate) createTemplateSubscriptions() error {
 	return nil
 }
 
-func (t *streamTemplate) processInboundTemplateMsg(_ *subscription, pc *client, subject, reply string, msg []byte) {
+func (t *streamTemplate) processInboundTemplateMsg(_ *subscription, pc *client, acc *Account, subject, reply string, msg []byte) {
 	if t == nil || t.jsa == nil {
 		return
 	}
@@ -1839,7 +1839,6 @@ func (t *streamTemplate) processInboundTemplateMsg(_ *subscription, pc *client,
 		jsa.mu.Unlock()
 		return
 	}
-	acc := jsa.account
 	jsa.mu.Unlock()
 
 	// Check if we are at the maximum and grab some variables.
@@ -1870,7 +1869,7 @@ func (t *streamTemplate) processInboundTemplateMsg(_ *subscription, pc *client,
 	}
 
 	// Process this message directly by invoking mset.
-	mset.processInboundJetStreamMsg(nil, pc, subject, reply, msg)
+	mset.processInboundJetStreamMsg(nil, pc, acc, subject, reply, msg)
 }
 
 // lookupStreamTemplate looks up the names stream template.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index e78e131d..b830727f 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -595,14 +595,13 @@ type JSApiStreamTemplateNamesResponse struct {
 
 const JSApiStreamTemplateNamesResponseType = "io.nats.jetstream.api.v1.stream_template_names_response"
 
-func (js *jetStream) apiDispatch(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, subject, reply string, rmsg []byte) {
 	hdr, _ := c.msgParts(rmsg)
 	if len(getHeader(ClientInfoHdr, hdr)) == 0 {
 		return
 	}
 	js.mu.RLock()
-	s := js.srv
-	rr := js.apiSubs.Match(subject)
+	s, rr := js.srv, js.apiSubs.Match(subject)
 	js.mu.RUnlock()
 
 	// Shortcircuit.
@@ -620,7 +619,7 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, subject, reply st
 
 	// If this is directly from a client connection ok to do in place.
 	if c.kind == CLIENT {
-		jsub.icb(sub, c, subject, reply, rmsg)
+		jsub.icb(sub, c, acc, subject, reply, rmsg)
 		return
 	}
 
@@ -652,7 +651,7 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, subject, reply st
 
 	// Dispatch the API call to its own Go routine.
 	go func() {
-		jsub.icb(sub, client, subject, reply, rmsg)
+		jsub.icb(sub, client, acc, subject, reply, rmsg)
 		atomic.AddInt64(&js.apiCalls, -1)
 	}()
 }
@@ -825,7 +824,7 @@ func (a *Account) checkJetStream() (enabled, shouldError bool) {
 }
 
 // Request for current usage and limits for this account.
-func (s *Server) jsAccountInfoRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsAccountInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -885,7 +884,7 @@ func consumerNameFromSubject(subject string) string {
 }
 
 // Request to create a new template.
-func (s *Server) jsTemplateCreateRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsTemplateCreateRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil {
 		return
 	}
@@ -940,7 +939,7 @@ func (s *Server) jsTemplateCreateRequest(sub *subscription, c *client, subject,
 }
 
 // Request for the list of all template names.
-func (s *Server) jsTemplateNamesRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsTemplateNamesRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil {
 		return
 	}
@@ -1004,7 +1003,7 @@ func (s *Server) jsTemplateNamesRequest(sub *subscription, c *client, subject, r
 }
 
 // Request for information about a stream template.
-func (s *Server) jsTemplateInfoRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsTemplateInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil {
 		return
 	}
@@ -1045,7 +1044,7 @@ func (s *Server) jsTemplateInfoRequest(sub *subscription, c *client, subject, re
 }
 
 // Request to delete a stream template.
-func (s *Server) jsTemplateDeleteRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsTemplateDeleteRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil {
 		return
 	}
@@ -1087,7 +1086,7 @@ func (s *Server) jsonResponse(v interface{}) string {
 }
 
 // Request to create a stream.
-func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, a *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1275,7 +1274,7 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, subject, re
 }
 
 // Request to update a stream.
-func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1358,7 +1357,7 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, subject, re
 }
 
 // Request for the list of all stream names.
-func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1482,7 +1481,7 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, subject, rep
 
 // Request for the list of all detailed stream info.
 // TODO(dlc) - combine with above long term
-func (s *Server) jsStreamListRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamListRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1572,7 +1571,7 @@ func (s *Server) jsStreamListRequest(sub *subscription, c *client, subject, repl
 }
 
 // Request for information about a stream.
-func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1684,7 +1683,7 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 }
 
 // Request to have a stream leader stepdown.
-func (s *Server) jsStreamLeaderStepDownRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamLeaderStepDownRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1773,7 +1772,7 @@ func (s *Server) jsStreamLeaderStepDownRequest(sub *subscription, c *client, sub
 }
 
 // Request to have a consumer leader stepdown.
-func (s *Server) jsConsumerLeaderStepDownRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsConsumerLeaderStepDownRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1872,7 +1871,7 @@ func (s *Server) jsConsumerLeaderStepDownRequest(sub *subscription, c *client, s
 }
 
 // Request to remove a peer from a clustered stream.
-func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -1974,7 +1973,7 @@ func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, subject
 }
 
 // Request to have the metaleader remove a peer from the system.
-func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2042,7 +2041,7 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, subje
 
 // Request to have the meta leader stepdown.
 // These will only be received the the meta leaders, so less checking needed.
-func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2136,7 +2135,7 @@ func isEmptyRequest(req []byte) bool {
 }
 
 // Request to delete a stream.
-func (s *Server) jsStreamDeleteRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamDeleteRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2204,7 +2203,7 @@ func (s *Server) jsStreamDeleteRequest(sub *subscription, c *client, subject, re
 
 // Request to delete a message.
 // This expects a stream sequence number as the msg body.
-func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2313,7 +2312,7 @@ func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, subject, reply
 }
 
 // Request to get a raw stream message.
-func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsMsgGetRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2433,7 +2432,7 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply st
 }
 
 // Request to purge a stream.
-func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2544,7 +2543,7 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, rep
 }
 
 // Request to restore a stream.
-func (s *Server) jsStreamRestoreRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamRestoreRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamIsLeader() {
 		return
 	}
@@ -2653,7 +2652,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 	var total int
 
 	// FIXM(dlc) - Probably take out of network path eventually due to disk I/O?
-	processChunk := func(sub *subscription, c *client, subject, reply string, msg []byte) {
+	processChunk := func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 		// We require reply subjects to communicate back failures, flow etc. If they do not have one log and cancel.
 		if reply == _EMPTY_ {
 			sub.client.processUnsub(sub.sid)
@@ -2801,7 +2800,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 }
 
 // Process a snapshot request.
-func (s *Server) jsStreamSnapshotRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsStreamSnapshotRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -2959,7 +2958,7 @@ func (s *Server) streamSnapshot(ci *ClientInfo, acc *Account, mset *stream, sr *
 
 	// We will place sequence number and size of chunk sent in the reply.
 	ackSubj := fmt.Sprintf(jsSnapshotAckT, mset.name(), nuid.Next())
-	ackSub, _ := mset.subscribeInternalUnlocked(ackSubj+".>", func(_ *subscription, _ *client, subject, _ string, _ []byte) {
+	ackSub, _ := mset.subscribeInternalUnlocked(ackSubj+".>", func(_ *subscription, _ *client, _ *Account, subject, _ string, _ []byte) {
 		cs, _ := strconv.Atoi(tokenAt(subject, 6))
 		// This is very crude and simple, but ok for now.
 		// This only matters when sending multiple chunks.
@@ -3006,16 +3005,17 @@ done:
 }
 
 // Request to create a durable consumer.
-func (s *Server) jsDurableCreateRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
-	s.jsConsumerCreate(sub, c, subject, reply, msg, true)
+func (s *Server) jsDurableCreateRequest(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
+	s.jsConsumerCreate(sub, c, acc, subject, reply, msg, true)
 }
 
 // Request to create a consumer.
-func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
-	s.jsConsumerCreate(sub, c, subject, reply, msg, false)
+func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
+	s.jsConsumerCreate(sub, c, acc, subject, reply, msg, false)
 }
 
-func (s *Server) jsConsumerCreate(sub *subscription, c *client, subject, reply string, rmsg []byte, expectDurable bool) {
+func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subject, reply string, rmsg []byte, expectDurable bool) {
+
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -3133,7 +3133,7 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, subject, reply s
 }
 
 // Request for the list of all consumer names.
-func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -3253,7 +3253,7 @@ func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, subject, r
 }
 
 // Request for the list of all detailed consumer information.
-func (s *Server) jsConsumerListRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -3346,7 +3346,7 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, subject, re
 }
 
 // Request for information about an consumer.
-func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
@@ -3460,7 +3460,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
 }
 
 // Request to delete an Consumer.
-func (s *Server) jsConsumerDeleteRequest(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+func (s *Server) jsConsumerDeleteRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 896043cf..21c06150 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -3071,7 +3071,7 @@ type streamAssignmentResult struct {
 
 // Process error results of stream and consumer assignments.
 // Success will be handled by stream leader.
-func (js *jetStream) processStreamAssignmentResults(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (js *jetStream) processStreamAssignmentResults(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	var result streamAssignmentResult
 	if err := json.Unmarshal(msg, &result); err != nil {
 		// TODO(dlc) - log
@@ -3111,7 +3111,7 @@ func (js *jetStream) processStreamAssignmentResults(sub *subscription, c *client
 	}
 }
 
-func (js *jetStream) processConsumerAssignmentResults(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (js *jetStream) processConsumerAssignmentResults(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	var result consumerAssignmentResult
 	if err := json.Unmarshal(msg, &result); err != nil {
 		// TODO(dlc) - log
@@ -3629,7 +3629,7 @@ func (s *Server) jsClusteredStreamListRequest(acc *Account, ci *ClientInfo, offs
 	rc := make(chan *StreamInfo, len(streams))
 
 	// Store our handler.
-	s.sys.replies[inbox] = func(sub *subscription, _ *client, subject, _ string, msg []byte) {
+	s.sys.replies[inbox] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 		var si StreamInfo
 		if err := json.Unmarshal(msg, &si); err != nil {
 			s.Warnf("Error unmarshaling clustered stream info response:%v", err)
@@ -3760,7 +3760,7 @@ func (s *Server) jsClusteredConsumerListRequest(acc *Account, ci *ClientInfo, of
 	rc := make(chan *ConsumerInfo, len(consumers))
 
 	// Store our handler.
-	s.sys.replies[inbox] = func(sub *subscription, _ *client, subject, _ string, msg []byte) {
+	s.sys.replies[inbox] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 		var ci ConsumerInfo
 		if err := json.Unmarshal(msg, &ci); err != nil {
 			s.Warnf("Error unmarshaling clustered consumer info response:%v", err)
@@ -4510,7 +4510,7 @@ RETRY:
 
 	// Send our catchup request here.
 	reply := syncReplySubject()
-	sub, err = s.sysSubscribe(reply, func(_ *subscription, _ *client, _, reply string, msg []byte) {
+	sub, err = s.sysSubscribe(reply, func(_ *subscription, _ *client, _ *Account, _, reply string, msg []byte) {
 		// Make copies - https://github.com/go101/go101/wiki
 		// TODO(dlc) - Since we are using a buffer from the inbound client/route.
 		select {
@@ -4606,7 +4606,7 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 	return seq, nil
 }
 
-func (mset *stream) handleClusterSyncRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (mset *stream) handleClusterSyncRequest(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	var sreq streamSyncRequest
 	if err := json.Unmarshal(msg, &sreq); err != nil {
 		// Log error.
@@ -4691,7 +4691,7 @@ func (mset *stream) checkClusterInfo(si *StreamInfo) {
 	}
 }
 
-func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	mset.mu.RLock()
 	sysc, js, sa, config := mset.sysc, mset.srv.js, mset.sa, mset.cfg
 	stype := mset.cfg.Storage
@@ -4748,7 +4748,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 
 	// Setup ackReply for flow control.
 	ackReply := syncAckSubject()
-	ackSub, _ := s.sysSubscribe(ackReply, func(sub *subscription, c *client, subject, reply string, msg []byte) {
+	ackSub, _ := s.sysSubscribe(ackReply, func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 		sz := ackReplySize(subject)
 		atomic.AddInt64(&outb, -sz)
 		atomic.AddInt32(&outm, -1)
diff --git a/server/mqtt.go b/server/mqtt.go
index 4014d2ea..7c7937cf 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -1,4 +1,4 @@
-// Copyright 2020 The NATS Authors
+// Copyright 2020-2021 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
@@ -1220,7 +1220,7 @@ func isErrorOtherThan(err error, id ErrorIdentifier) bool {
 // Process JS API replies.
 //
 // Can run from various go routines (consumer's loop, system send loop, etc..).
-func (as *mqttAccountSessionManager) processJSAPIReplies(_ *subscription, pc *client, subject, _ string, msg []byte) {
+func (as *mqttAccountSessionManager) processJSAPIReplies(_ *subscription, pc *client, _ *Account, subject, _ string, msg []byte) {
 	token := tokenAt(subject, mqttJSATokenPos)
 	if token == _EMPTY_ {
 		return
@@ -1284,7 +1284,7 @@ func (as *mqttAccountSessionManager) processJSAPIReplies(_ *subscription, pc *cl
 //
 // Run from various go routines (JS consumer, etc..).
 // No lock held on entry.
-func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *client, subject, reply string, rmsg []byte) {
+func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	_, msg := c.msgParts(rmsg)
 	rm := &mqttRetainedMsg{}
 	if err := json.Unmarshal(msg, rm); err != nil {
@@ -1310,7 +1310,7 @@ func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *clie
 	}
 }
 
-func (as *mqttAccountSessionManager) processRetainedMsgDel(_ *subscription, c *client, subject, reply string, rmsg []byte) {
+func (as *mqttAccountSessionManager) processRetainedMsgDel(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	idHash := tokenAt(subject, 3)
 	if idHash == _EMPTY_ || idHash == as.jsa.id {
 		return
@@ -1334,7 +1334,7 @@ func (as *mqttAccountSessionManager) processRetainedMsgDel(_ *subscription, c *c
 //
 // Can run from various go routines (system send loop, etc..).
 // No lock held on entry.
-func (as *mqttAccountSessionManager) processSessionPersist(_ *subscription, pc *client, subject, _ string, rmsg []byte) {
+func (as *mqttAccountSessionManager) processSessionPersist(_ *subscription, pc *client, _ *Account, subject, _ string, rmsg []byte) {
 	// Ignore our own responses here (they are handled elsewhere)
 	if tokenAt(subject, mqttJSAIdTokenPos) == as.jsa.id {
 		return
@@ -2500,7 +2500,7 @@ CHECK:
 			asm.mu.Lock()
 			asm.addSessToFlappers(cp.clientID)
 			asm.mu.Unlock()
-			c.Warnf("Replacing old client %q since both have the same client ID %q", ec.String(), cp.clientID)
+			c.Warnf("Replacing old client %q since both have the same client ID %q", ec, cp.clientID)
 			// Close old client in separate go routine
 			go ec.closeConnection(DuplicateClientID)
 		}
@@ -3047,7 +3047,7 @@ func mqttSubscribeTrace(pi uint16, filters []*mqttFilter) string {
 // message and this is the callback for a QoS1 subscription because in
 // that case, it will be handled by the other callback. This avoid getting
 // duplicate deliveries.
-func mqttDeliverMsgCbQos0(sub *subscription, pc *client, subject, _ string, rmsg []byte) {
+func mqttDeliverMsgCbQos0(sub *subscription, pc *client, _ *Account, subject, _ string, rmsg []byte) {
 	if pc.kind == JETSTREAM {
 		return
 	}
@@ -3110,7 +3110,7 @@ func mqttDeliverMsgCbQos0(sub *subscription, pc *client, subject, _ string, rmsg
 // associated with the JS durable consumer), but in cluster mode, this can be coming
 // from a route, gw, etc... We make sure that if this is the case, the message contains
 // a NATS/MQTT header that indicates that this is a published QoS1 message.
-func mqttDeliverMsgCbQos1(sub *subscription, pc *client, subject, reply string, rmsg []byte) {
+func mqttDeliverMsgCbQos1(sub *subscription, pc *client, _ *Account, subject, reply string, rmsg []byte) {
 	var retained bool
 
 	// Message on foo.bar is stored under $MQTT.msgs.foo.bar, so the subject has to be
diff --git a/server/raft.go b/server/raft.go
index c6fd43e9..70382c47 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -1651,7 +1651,7 @@ func (n *raft) decodeAppendEntryResponse(msg []byte) *appendEntryResponse {
 }
 
 // Called when a remove peer proposal has been forwarded
-func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _, reply string, msg []byte) {
+func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _ *Account, _, reply string, msg []byte) {
 	n.debug("Received forwarded remove peer proposal: %q", msg)
 
 	if !n.Leader() {
@@ -1682,7 +1682,7 @@ func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _
 }
 
 // Called when a peer has forwarded a proposal.
-func (n *raft) handleForwardedProposal(sub *subscription, c *client, _, reply string, msg []byte) {
+func (n *raft) handleForwardedProposal(sub *subscription, c *client, _ *Account, _, reply string, msg []byte) {
 	if !n.Leader() {
 		n.debug("Ignoring forwarded proposal, not leader")
 		return
@@ -2334,7 +2334,7 @@ func (n *raft) runAsCandidate() {
 }
 
 // handleAppendEntry handles an append entry from the wire.
-func (n *raft) handleAppendEntry(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (n *raft) handleAppendEntry(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	if n.outOfResources() {
 		n.debug("AppendEntry not processing inbound, no resources")
 		return
@@ -2733,7 +2733,7 @@ func (n *raft) processAppendEntryResponse(ar *appendEntryResponse) {
 }
 
 // handleAppendEntryResponse processes responses to append entries.
-func (n *raft) handleAppendEntryResponse(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (n *raft) handleAppendEntryResponse(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	// Ignore if not the leader.
 	if !n.Leader() {
 		return
@@ -3072,7 +3072,7 @@ func (n *raft) decodeVoteResponse(msg []byte) *voteResponse {
 	return vr
 }
 
-func (n *raft) handleVoteResponse(sub *subscription, c *client, _, reply string, msg []byte) {
+func (n *raft) handleVoteResponse(sub *subscription, c *client, _ *Account, _, reply string, msg []byte) {
 	vr := n.decodeVoteResponse(msg)
 	n.debug("Received a voteResponse %+v", vr)
 	if vr == nil {
@@ -3137,7 +3137,7 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	return nil
 }
 
-func (n *raft) handleVoteRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
+func (n *raft) handleVoteRequest(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	vr := n.decodeVoteRequest(msg, reply)
 	if vr == nil {
 		n.error("Received malformed vote request for %q", n.group)
diff --git a/server/stream.go b/server/stream.go
index d3866f01..c7e54ef6 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1530,7 +1530,7 @@ func (mset *stream) setupMirrorConsumer() error {
 
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
 	reply := infoReplySubject()
-	crSub, _ := mset.subscribeInternal(reply, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+	crSub, _ := mset.subscribeInternal(reply, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 		mset.unsubscribeUnlocked(sub)
 		_, msg := c.msgParts(rmsg)
 
@@ -1582,7 +1582,7 @@ func (mset *stream) setupMirrorConsumer() error {
 				msgs := mset.mirror.msgs
 
 				// Process inbound mirror messages from the wire.
-				sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+				sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 					hdr, msg := c.msgParts(append(rmsg[:0:0], rmsg...)) // Need to copy.
 					mset.queueInbound(msgs, subject, reply, hdr, msg)
 				})
@@ -1732,7 +1732,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
 	reply := infoReplySubject()
-	crSub, _ := mset.subscribeInternal(reply, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+	crSub, _ := mset.subscribeInternal(reply, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 		mset.unsubscribe(sub)
 		_, msg := c.msgParts(rmsg)
 		var ccr JSApiConsumerCreateResponse
@@ -1771,7 +1771,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 					// Capture consumer name.
 					si.cname = ccr.ConsumerInfo.Name
 					// Now create sub to receive messages.
-					sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+					sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 						hdr, msg := c.msgParts(append(rmsg[:0:0], rmsg...)) // Need to copy.
 						mset.queueInbound(si.msgs, subject, reply, hdr, msg)
 					})
@@ -2512,7 +2512,7 @@ func (mset *stream) queueInboundMsg(subj, rply string, hdr, msg []byte) {
 }
 
 // processInboundJetStreamMsg handles processing messages bound for a stream.
-func (mset *stream) processInboundJetStreamMsg(_ *subscription, c *client, subject, reply string, rmsg []byte) {
+func (mset *stream) processInboundJetStreamMsg(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	mset.mu.RLock()
 	isLeader, isClustered := mset.isLeader(), mset.node != nil
 	mset.mu.RUnlock()


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 69da9a16..abdd0a7b 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -635,7 +635,7 @@ func TestSysSubscribeRace(t *testing.T) {
 
 	received := make(chan struct{})
 	// Create message callback handler.
-	cb := func(sub *subscription, producer *client, subject, reply string, msg []byte) {
+	cb := func(sub *subscription, producer *client, _ *Account, subject, reply string, msg []byte) {
 		select {
 		case received <- struct{}{}:
 		default:
@@ -682,7 +682,7 @@ func TestSystemAccountInternalSubscriptions(t *testing.T) {
 
 	received := make(chan *nats.Msg)
 	// Create message callback handler.
-	cb := func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+	cb := func(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 		copy := append([]byte(nil), msg...)
 		received <- &nats.Msg{Subject: subject, Reply: reply, Data: copy}
 	}
@@ -756,7 +756,7 @@ func TestSystemAccountConnectionUpdatesStopAfterNoLocal(t *testing.T) {
 
 	// Listen for updates to the new account connection activity.
 	received := make(chan *nats.Msg, 10)
-	cb := func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+	cb := func(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 		copy := append([]byte(nil), msg...)
 		received <- &nats.Msg{Subject: subject, Reply: reply, Data: copy}
 	}
@@ -1574,7 +1574,7 @@ func TestSystemAccountWithBadRemoteLatencyUpdate(t *testing.T) {
 		ReqId:   "_INBOX.22",
 	}
 	b, _ := json.Marshal(&rl)
-	s.remoteLatencyUpdate(nil, nil, "foo", "", b)
+	s.remoteLatencyUpdate(nil, nil, nil, "foo", _EMPTY_, b)
 }
 
 func TestSystemAccountWithGateways(t *testing.T) {
@@ -2132,7 +2132,7 @@ func TestConnectionUpdatesTimerProperlySet(t *testing.T) {
 
 	// Listen for HB updates...
 	count := int32(0)
-	cb := func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+	cb := func(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 		atomic.AddInt32(&count, 1)
 	}
 	subj := fmt.Sprintf(accConnsEventSubjOld, pub)
diff --git a/server/norace_test.go b/server/norace_test.go
index b763cfdb..9f0595f9 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1102,6 +1102,89 @@ func TestNoRaceJetStreamDeleteStreamManyConsumers(t *testing.T) {
 	mset.delete()
 }
 
+// We used to swap accounts on an inbound message when processing service imports.
+// Until JetStream this was kinda ok, but with JetStream we can have pull consumers
+// trying to access the clients account in another Go routine now which causes issues.
+// This is not limited to the case above, its just the one that exposed it.
+// This test is to show that issue and that the fix works, meaning we no longer swap c.acc.
+func TestNoRaceJetStreamServiceImportAccountSwapIssue(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	// Client based API
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo", "bar"},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sub, err := js.PullSubscribe("foo", "dlc")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	beforeSubs := s.NumSubscriptions()
+
+	// How long we want both sides to run.
+	timeout := time.Now().Add(3 * time.Second)
+	errs := make(chan error, 1)
+
+	// Publishing side, which will signal the consumer that is waiting and which will access c.acc. If publish
+	// operation runs concurrently we will catch c.acc being $SYS some of the time.
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		for time.Now().Before(timeout) {
+			// This will signal the delivery of the pull messages.
+			js.Publish("foo", []byte("Hello"))
+			// This will swap the account because of JetStream service import.
+			// We can get an error here with the bug or not.
+			if _, err := js.StreamInfo("TEST"); err != nil {
+				errs <- err
+				return
+			}
+		}
+		errs <- nil
+	}()
+
+	// Pull messages flow.
+	var received int
+	for time.Now().Before(timeout) {
+		if msgs, err := sub.Fetch(1, nats.MaxWait(200*time.Millisecond)); err == nil {
+			for _, m := range msgs {
+				received++
+				m.Ack()
+			}
+		} else {
+			break
+		}
+	}
+	// Wait on publisher Go routine and check for errors.
+	if err := <-errs; err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// Double check all received.
+	si, err := js.StreamInfo("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if int(si.State.Msgs) != received {
+		t.Fatalf("Expected to receive %d msgs, only got %d", si.State.Msgs, received)
+	}
+	// Now check for leaked subs from the fetch call above. That is what we first saw from the bug.
+	if afterSubs := s.NumSubscriptions(); afterSubs != beforeSubs {
+		t.Fatalf("Leaked subscriptions: %d before, %d after", beforeSubs, afterSubs)
+	}
+}
+
 func TestNoRaceJetStreamAPIStreamListPaging(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()