Repository: nats-io/nats-server
Pull request #3319 (2022-08-02 04:00:50)

Adding account purge operation
The new request is available for the system account.
The subject to send the request to is `$JS.API.ACCOUNT.PURGE.*`
With the name of the account to purge instead of the wildcard.

Also added directory cleanup code such that server do not
end up with empty streams directories and account dirs that
only contain streams

Also adding ACCOUNT to leaf node domain rewrite table

Addresses #3186 and #3306 by providing a way to
get rid of the streams for existing and non existing accounts


Patch:
diff --git a/server/const.go b/server/const.go
index 0168caf2..2970ed47 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-beta.20"
+	VERSION = "2.9.0-beta.21"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/jetstream.go b/server/jetstream.go
index b0b068dd..39173d9e 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -94,20 +94,22 @@ type JetStreamAPIStats struct {
 // This is for internal accounting for JetStream for this server.
 type jetStream struct {
 	// These are here first because of atomics on 32bit systems.
-	apiInflight    int64
-	apiTotal       int64
-	apiErrors      int64
-	memReserved    int64
-	storeReserved  int64
-	memUsed        int64
-	storeUsed      int64
-	clustered      int32
-	mu             sync.RWMutex
-	srv            *Server
-	config         JetStreamConfig
-	cluster        *jetStreamCluster
-	accounts       map[string]*jsAccount
-	apiSubs        *Sublist
+	apiInflight   int64
+	apiTotal      int64
+	apiErrors     int64
+	memReserved   int64
+	storeReserved int64
+	memUsed       int64
+	storeUsed     int64
+	clustered     int32
+	mu            sync.RWMutex
+	srv           *Server
+	config        JetStreamConfig
+	cluster       *jetStreamCluster
+	accounts      map[string]*jsAccount
+	apiSubs       *Sublist
+	// System level request to purge a stream move
+	accountPurge   *subscription
 	metaRecovering bool
 	standAlone     bool
 	disabled       bool
@@ -689,6 +691,21 @@ func (s *Server) configAllJetStreamAccounts() error {
 		return nil
 	}
 
+	if s.sys != nil {
+		// clustered stream removal will perform this cleanup as well
+		// this is mainly for initial cleanup
+		saccName := s.sys.account.Name
+		accStoreDirs, _ := ioutil.ReadDir(js.config.StoreDir)
+		for _, acc := range accStoreDirs {
+			if accName := acc.Name(); accName != saccName {
+				// no op if not empty
+				accDir := filepath.Join(js.config.StoreDir, accName)
+				os.Remove(filepath.Join(accDir, streamsDir))
+				os.Remove(accDir)
+			}
+		}
+	}
+
 	var jsAccounts []*Account
 	s.accounts.Range(func(k, v interface{}) bool {
 		jsAccounts = append(jsAccounts, v.(*Account))
@@ -741,6 +758,12 @@ func (js *jetStream) setJetStreamStandAlone(isStandAlone bool) {
 	js.mu.Lock()
 	defer js.mu.Unlock()
 	js.standAlone = isStandAlone
+
+	if isStandAlone {
+		js.accountPurge, _ = js.srv.systemSubscribe(JSApiAccountPurge, _EMPTY_, false, nil, js.srv.jsLeaderAccountPurgeRequest)
+	} else if js.accountPurge != nil {
+		js.srv.sysUnsubscribe(js.accountPurge)
+	}
 }
 
 // JetStreamEnabled reports if jetstream is enabled for this server.
@@ -872,14 +895,20 @@ func (s *Server) shutdownJetStream() {
 	var _a [512]*Account
 	accounts := _a[:0]
 
-	js.mu.RLock()
+	js.mu.Lock()
 	// Collect accounts.
 	for _, jsa := range js.accounts {
 		if a := jsa.acc(); a != nil {
 			accounts = append(accounts, a)
 		}
 	}
-	js.mu.RUnlock()
+	accPurgeSub := js.accountPurge
+	js.accountPurge = nil
+	js.mu.Unlock()
+
+	if accPurgeSub != nil {
+		s.sysUnsubscribe(accPurgeSub)
+	}
 
 	for _, a := range accounts {
 		a.removeJetStream()
@@ -1052,7 +1081,8 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 		// Just need to make sure we can write to the directory.
 		// Remove the directory will create later if needed.
 		os.RemoveAll(sdir)
-
+		// when empty remove parent directory, which may have been created as well
+		os.Remove(jsa.storeDir)
 	} else {
 		// Restore any state here.
 		s.Debugf("Recovering JetStream state for account %q", a.Name)
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index f9c4198e..800907c3 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -195,6 +195,12 @@ const (
 	// Will return JSON response.
 	JSApiRemoveServer = "$JS.API.SERVER.REMOVE"
 
+	// JSApiAccountPurge is the endpoint to purge the js content of an account
+	// Only works from system account.
+	// Will return JSON response.
+	JSApiAccountPurge  = "$JS.API.ACCOUNT.PURGE.*"
+	JSApiAccountPurgeT = "$JS.API.ACCOUNT.PURGE.%s"
+
 	// JSApiServerStreamMove is the endpoint to move streams off a server
 	// Only works from system account.
 	// Will return JSON response.
@@ -308,6 +314,7 @@ func generateJSMappingTable(domain string) map[string]string {
 		"DIRECT.>":   "$JS.API.DIRECT.>",
 		"META.>":     "$JS.API.META.>",
 		"SERVER.>":   "$JS.API.SERVER.>",
+		"ACCOUNT.>":  "$JS.API.ACCOUNT.>",
 		"$KV.>":      "$KV.>",
 		"$OBJ.>":     "$OBJ.>",
 	} {
@@ -592,6 +599,14 @@ type JSApiMetaServerStreamMoveRequest struct {
 	Tags []string `json:"tags,omitempty"`
 }
 
+const JSApiAccountPurgeResponseType = "io.nats.jetstream.api.v1.account_purge_response"
+
+// JSApiAccountPurgeResponse is the response to a purge request in the meta group.
+type JSApiAccountPurgeResponse struct {
+	ApiResponse
+	Initiated bool `json:"initiated,omitempty"`
+}
+
 // JSApiMsgGetRequest get a message request.
 type JSApiMsgGetRequest struct {
 	Seq     uint64 `json:"seq,omitempty"`
@@ -2456,6 +2471,100 @@ func (s *Server) jsLeaderServerStreamCancelMoveRequest(sub *subscription, c *cli
 	s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
 }
 
+// Request to have an account purged
+func (s *Server) jsLeaderAccountPurgeRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+	if c == nil || !s.JetStreamEnabled() {
+		return
+	}
+
+	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	}
+
+	js := s.getJetStream()
+	if js == nil {
+		return
+	}
+
+	accName := tokenAt(subject, 5)
+
+	var resp = JSApiAccountPurgeResponse{ApiResponse: ApiResponse{Type: JSApiAccountPurgeResponseType}}
+
+	if !s.JetStreamIsClustered() {
+		var streams []*stream
+		var ac *Account
+		if ac, err = s.lookupAccount(accName); err == nil && ac != nil {
+			streams = ac.streams()
+		}
+
+		s.Noticef("Purge request for account %s (streams: %d, hasAccount: %t)",
+			accName, len(streams), ac != nil)
+
+		for _, mset := range streams {
+			err := mset.delete()
+			if err != nil {
+				resp.Error = NewJSStreamDeleteError(err)
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+				return
+			}
+		}
+		if err := os.RemoveAll(filepath.Join(js.config.StoreDir, accName)); err != nil {
+			resp.Error = NewJSStreamGeneralError(err)
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		resp.Initiated = true
+		s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	_, cc := s.getJetStreamCluster()
+	if cc == nil || cc.meta == nil || !cc.isLeader() {
+		return
+	}
+
+	if js.isMetaRecovering() {
+		// While in recovery mode, the data structures are not fully initialized
+		resp.Error = NewJSClusterNotAvailError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	rm := recoveryRemovals{
+		streams:   make(map[string]*streamAssignment),
+		consumers: make(map[string]*consumerAssignment),
+	}
+
+	js.mu.Lock()
+	streams, hasAccount := cc.streams[accName]
+	if hasAccount {
+		for _, s := range streams {
+			key := accName + ":" + s.Config.Name
+			rm.streams[key] = s.copyGroup()
+			for _, c := range s.consumers {
+				key := accName + ":" + s.Config.Name + ":" + c.Config.Durable
+				rm.consumers[key] = c.copyGroup()
+			}
+		}
+	}
+	js.mu.Unlock()
+
+	s.Noticef("Purge request for account %s (streams: %d, consumer: %d, hasAccount: %t)",
+		accName, len(rm.streams), len(rm.consumers), hasAccount)
+
+	for _, ca := range rm.consumers {
+		cc.meta.Propose(encodeDeleteConsumerAssignment(ca))
+	}
+	for _, sa := range rm.streams {
+		cc.meta.Propose(encodeDeleteStreamAssignment(sa))
+	}
+	resp.Initiated = true
+
+	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+}
+
 // 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, _ *Account, subject, reply string, rmsg []byte) {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index aa1842d7..bc5d23f0 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"os"
 	"path/filepath"
 	"reflect"
 	"sort"
@@ -914,12 +915,12 @@ func (js *jetStream) monitorCluster() {
 					js.clearMetaRecovering()
 
 					// Process any removes that are still valid after recovery.
-					for _, sa := range rm.streams {
-						js.processStreamRemoval(sa)
-					}
 					for _, ca := range rm.consumers {
 						js.processConsumerRemoval(ca)
 					}
+					for _, sa := range rm.streams {
+						js.processStreamRemoval(sa)
+					}
 					// Clear.
 					rm = nil
 					s.Debugf("Recovered JetStream cluster metadata")
@@ -1077,16 +1078,15 @@ func (js *jetStream) metaSnapshot() []byte {
 }
 
 func (js *jetStream) applyMetaSnapshot(buf []byte) error {
-	if len(buf) == 0 {
-		return nil
-	}
-	jse, err := s2.Decode(nil, buf)
-	if err != nil {
-		return err
-	}
 	var wsas []writeableStreamAssignment
-	if err = json.Unmarshal(jse, &wsas); err != nil {
-		return err
+	if len(buf) != 0 {
+		jse, err := s2.Decode(nil, buf)
+		if err != nil {
+			return err
+		}
+		if err = json.Unmarshal(jse, &wsas); err != nil {
+			return err
+		}
 	}
 
 	// Build our new version here outside of js.
@@ -1146,6 +1146,8 @@ func (js *jetStream) applyMetaSnapshot(buf []byte) error {
 		}
 	}
 	isRecovering := js.metaRecovering
+	storeDir := js.config.StoreDir
+	srv := js.srv
 	js.mu.Unlock()
 
 	// Do removals first.
@@ -1183,6 +1185,20 @@ func (js *jetStream) applyMetaSnapshot(buf []byte) error {
 		js.processConsumerAssignment(ca)
 	}
 
+	// Delete directories not part of snapshot
+	sysAcc := srv.SystemAccount().GetName()
+	if de, err := os.ReadDir(storeDir); err == nil {
+		for _, entry := range de {
+			acc := entry.Name()
+			if acc == sysAcc {
+				continue
+			}
+			if _, found := streams[acc]; !found {
+				os.RemoveAll(filepath.Join(storeDir, acc))
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -2573,6 +2589,23 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
 		return false
 	}
 
+	js.mu.Lock()
+	accStreams := cc.streams[accName]
+	if accStreams == nil {
+		accStreams = make(map[string]*streamAssignment)
+	} else if osa := accStreams[stream]; osa != nil && osa != sa {
+		// Copy over private existing state from former SA.
+		sa.Group.node = osa.Group.node
+		sa.consumers = osa.consumers
+		sa.responded = osa.responded
+		sa.err = osa.err
+	}
+
+	// Update our state.
+	accStreams[stream] = sa
+	cc.streams[accName] = accStreams
+	js.mu.Unlock()
+
 	acc, err := s.LookupAccount(accName)
 	if err != nil {
 		ll := fmt.Sprintf("Account [%s] lookup for stream create failed: %v", accName, err)
@@ -2592,23 +2625,6 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
 		return false
 	}
 
-	js.mu.Lock()
-	accStreams := cc.streams[acc.Name]
-	if accStreams == nil {
-		accStreams = make(map[string]*streamAssignment)
-	} else if osa := accStreams[stream]; osa != nil && osa != sa {
-		// Copy over private existing state from former SA.
-		sa.Group.node = osa.Group.node
-		sa.consumers = osa.consumers
-		sa.responded = osa.responded
-		sa.err = osa.err
-	}
-
-	// Update our state.
-	accStreams[stream] = sa
-	cc.streams[acc.Name] = accStreams
-	js.mu.Unlock()
-
 	var didRemove bool
 
 	// Check if this is for us..
@@ -2651,11 +2667,7 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 		return
 	}
 
-	acc, err := s.LookupAccount(sa.Client.serviceAccount())
-	if err != nil {
-		// TODO(dlc) - log error
-		return
-	}
+	accName := sa.Client.serviceAccount()
 	stream := sa.Config.Name
 
 	js.mu.Lock()
@@ -2670,7 +2682,7 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 		isMember = sa.Group.isMember(ourID)
 	}
 
-	accStreams := cc.streams[acc.Name]
+	accStreams := cc.streams[accName]
 	if accStreams == nil {
 		js.mu.Unlock()
 		return
@@ -2693,7 +2705,7 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 
 	// Update our state.
 	accStreams[stream] = sa
-	cc.streams[acc.Name] = accStreams
+	cc.streams[accName] = accStreams
 
 	// Make sure we respond if we are a member.
 	if isMember {
@@ -2704,6 +2716,12 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 	}
 	js.mu.Unlock()
 
+	acc, err := s.LookupAccount(accName)
+	if err != nil {
+		s.Warnf("Update Stream Account %s, error on lookup: %v", accName, err)
+		return
+	}
+
 	// Check if this is for us..
 	if isMember {
 		js.processClusterUpdateStream(acc, osa, sa)
@@ -3050,7 +3068,8 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	}
 	js.mu.RLock()
 	s := js.srv
-	hadLeader := sa.Group.node == nil || sa.Group.node.GroupLeader() != noLeader
+	node := sa.Group.node
+	hadLeader := node == nil || node.GroupLeader() != noLeader
 	offline := s.allPeersOffline(sa.Group)
 	var isMetaLeader bool
 	if cc := js.cluster; cc != nil {
@@ -3058,6 +3077,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	}
 	js.mu.RUnlock()
 
+	stopped := false
 	var resp = JSApiStreamDeleteResponse{ApiResponse: ApiResponse{Type: JSApiStreamDeleteResponseType}}
 	var err error
 	var acc *Account
@@ -3066,13 +3086,36 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	if acc, _ = s.LookupAccount(sa.Client.serviceAccount()); acc != nil {
 		if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
 			err = mset.stop(true, wasLeader)
+			stopped = true
 		}
 	}
 
 	// Always delete the node if present.
-	if sa.Group.node != nil {
-		sa.Group.node.Delete()
+	if node != nil {
+		node.Delete()
+	}
+
+	// This is a stop gap cleanup in case
+	// 1) the account does not exist (and mset couldn't be stopped) and/or
+	// 2) node was nil (and couldn't be deleted)
+	if !stopped || node == nil {
+		if sacc := s.SystemAccount(); sacc != nil {
+			os.RemoveAll(filepath.Join(js.config.StoreDir, sacc.GetName(), defaultStoreDirName, sa.Group.Name))
+			// cleanup dependent consumer groups
+			if !stopped {
+				for _, ca := range sa.consumers {
+					os.RemoveAll(filepath.Join(js.config.StoreDir, sacc.GetName(), defaultStoreDirName, ca.Group.Name))
+				}
+			}
+		}
 	}
+	accDir := filepath.Join(js.config.StoreDir, sa.Client.serviceAccount())
+	streamDir := filepath.Join(accDir, streamsDir)
+	os.RemoveAll(filepath.Join(streamDir, sa.Config.Name))
+
+	// no op if not empty
+	os.Remove(streamDir)
+	os.Remove(accDir)
 
 	// Normally we want only the leader to respond here, but if we had no leader then all members will respond to make
 	// sure we get feedback to the user.
@@ -3083,6 +3126,11 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 		}
 	}
 
+	// Do not respond if the account does not exist any longer
+	if acc == nil {
+		return
+	}
+
 	if err != nil {
 		resp.Error = NewJSStreamGeneralError(err, Unless(err))
 		s.sendAPIErrResponse(sa.Client, acc, sa.Subject, sa.Reply, _EMPTY_, s.jsonResponse(resp))
@@ -3112,26 +3160,6 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		return
 	}
 
-	acc, err := s.LookupAccount(accName)
-	if err != nil {
-		ll := fmt.Sprintf("Account [%s] lookup for consumer create failed: %v", accName, err)
-		if isMember {
-			// If we can not lookup the account and we are a member, send this result back to the metacontroller leader.
-			result := &consumerAssignmentResult{
-				Account:  accName,
-				Stream:   stream,
-				Consumer: consumerName,
-				Response: &JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}},
-			}
-			result.Response.Error = NewJSNoAccountError()
-			s.sendInternalMsgLocked(consumerAssignmentSubj, _EMPTY_, nil, result)
-			s.Warnf(ll)
-		} else {
-			s.Debugf(ll)
-		}
-		return
-	}
-
 	sa := js.streamAssignment(accName, stream)
 	if sa == nil {
 		s.Debugf("Consumer create failed, could not locate stream '%s > %s'", accName, stream)
@@ -3166,6 +3194,26 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	sa.consumers[ca.Name] = ca
 	js.mu.Unlock()
 
+	acc, err := s.LookupAccount(accName)
+	if err != nil {
+		ll := fmt.Sprintf("Account [%s] lookup for consumer create failed: %v", accName, err)
+		if isMember {
+			// If we can not lookup the account and we are a member, send this result back to the metacontroller leader.
+			result := &consumerAssignmentResult{
+				Account:  accName,
+				Stream:   stream,
+				Consumer: consumerName,
+				Response: &JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}},
+			}
+			result.Response.Error = NewJSNoAccountError()
+			s.sendInternalMsgLocked(consumerAssignmentSubj, _EMPTY_, nil, result)
+			s.Warnf(ll)
+		} else {
+			s.Debugf(ll)
+		}
+		return
+	}
+
 	// Check if this is for us..
 	if isMember {
 		js.processClusterCreateConsumer(ca, state, wasExisting)
@@ -3474,6 +3522,7 @@ func (js *jetStream) processClusterDeleteConsumer(ca *consumerAssignment, isMemb
 	}
 	js.mu.RLock()
 	s := js.srv
+	node := ca.Group.node
 	offline := s.allPeersOffline(ca.Group)
 	var isMetaLeader bool
 	if cc := js.cluster; cc != nil {
@@ -3481,6 +3530,7 @@ func (js *jetStream) processClusterDeleteConsumer(ca *consumerAssignment, isMemb
 	}
 	js.mu.RUnlock()
 
+	stopped := false
 	var resp = JSApiConsumerDeleteResponse{ApiResponse: ApiResponse{Type: JSApiConsumerDeleteResponseType}}
 	var err error
 	var acc *Account
@@ -3490,13 +3540,23 @@ func (js *jetStream) processClusterDeleteConsumer(ca *consumerAssignment, isMemb
 		if mset, _ := acc.lookupStream(ca.Stream); mset != nil {
 			if o := mset.lookupConsumer(ca.Name); o != nil {
 				err = o.stopWithFlags(true, false, true, wasLeader)
+				stopped = true
 			}
 		}
 	}
 
 	// Always delete the node if present.
-	if ca.Group.node != nil {
-		ca.Group.node.Delete()
+	if node != nil {
+		node.Delete()
+	}
+
+	// This is a stop gap cleanup in case
+	// 1) the account does not exist (and mset consumer couldn't be stopped) and/or
+	// 2) node was nil (and couldn't be deleted)
+	if !stopped || node == nil {
+		if sacc := s.SystemAccount(); sacc != nil {
+			os.RemoveAll(filepath.Join(js.config.StoreDir, sacc.GetName(), defaultStoreDirName, ca.Group.Name))
+		}
 	}
 
 	if !wasLeader || ca.Reply == _EMPTY_ {
@@ -3505,6 +3565,11 @@ func (js *jetStream) processClusterDeleteConsumer(ca *consumerAssignment, isMemb
 		}
 	}
 
+	// Do not respond if the account does not exist any longer
+	if acc == nil {
+		return
+	}
+
 	if err != nil {
 		resp.Error = NewJSStreamNotFoundError(Unless(err))
 		s.sendAPIErrResponse(ca.Client, acc, ca.Subject, ca.Reply, _EMPTY_, s.jsonResponse(resp))
@@ -4277,6 +4342,9 @@ func (js *jetStream) startUpdatesSub() {
 	if cc.peerStreamCancelMove == nil {
 		cc.peerStreamCancelMove, _ = s.systemSubscribe(JSApiServerStreamCancelMove, _EMPTY_, false, c, s.jsLeaderServerStreamCancelMoveRequest)
 	}
+	if js.accountPurge == nil {
+		js.accountPurge, _ = s.systemSubscribe(JSApiAccountPurge, _EMPTY_, false, c, s.jsLeaderAccountPurgeRequest)
+	}
 }
 
 // Lock should be held.
@@ -4306,6 +4374,10 @@ func (js *jetStream) stopUpdatesSub() {
 		cc.s.sysUnsubscribe(cc.peerStreamCancelMove)
 		cc.peerStreamCancelMove = nil
 	}
+	if js.accountPurge != nil {
+		cc.s.sysUnsubscribe(js.accountPurge)
+		js.accountPurge = nil
+	}
 }
 
 func (js *jetStream) processLeaderChange(isLeader bool) {
diff --git a/server/leafnode.go b/server/leafnode.go
index d2982ed3..5f906b05 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -495,12 +495,13 @@ func (s *Server) connectToRemoteLeafNode(remote *leafNodeCfg, firstConnect bool)
 			if url != rURL.Host {
 				ipStr = fmt.Sprintf(" (%s)", url)
 			}
-			if s.isLeafNodeEnabled() {
-				s.Debugf("Trying to connect as leafnode to remote server on %q%s", rURL.Host, ipStr)
-				conn, err = natsDialTimeout("tcp", url, dialTimeout)
-			} else {
+			// Some test may want to disable remotes from connecting
+			if s.isLeafConnectDisabled() {
 				s.Debugf("Will not attempt to connect to remote server on %q%s, leafnodes currently disabled", rURL.Host, ipStr)
 				err = ErrLeafNodeDisabled
+			} else {
+				s.Debugf("Trying to connect as leafnode to remote server on %q%s", rURL.Host, ipStr)
+				conn, err = natsDialTimeout("tcp", url, dialTimeout)
 			}
 		}
 		if err != nil {
@@ -569,10 +570,10 @@ func (s *Server) checkJetStreamMigrate(remote *leafNodeCfg) {
 }
 
 // Helper for checking.
-func (s *Server) isLeafNodeEnabled() bool {
+func (s *Server) isLeafConnectDisabled() bool {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
-	return s.leafNodeEnabled
+	return s.leafDisableConnect
 }
 
 // Save off the tlsName for when we use TLS and mix hostnames and IPs. IPs usually
@@ -706,8 +707,24 @@ func (c *client) sendLeafConnect(clusterName string, tlsRequired, headers bool)
 		DenyPub:   c.leaf.remote.DenyImports,
 	}
 
-	// Check for credentials first, that will take precedence..
-	if creds := c.leaf.remote.Credentials; creds != _EMPTY_ {
+	// If a signature callback is specified, this takes precedence over anything else.
+	if cb := c.leaf.remote.SignatureCB; cb != nil {
+		nonce := c.nonce
+		c.mu.Unlock()
+		jwt, sigraw, err := cb(nonce)
+		c.mu.Lock()
+		if err == nil && c.isClosed() {
+			err = ErrConnectionClosed
+		}
+		if err != nil {
+			c.Errorf("Error signing the nonce: %v", err)
+			return err
+		}
+		sig := base64.RawURLEncoding.EncodeToString(sigraw)
+		cinfo.JWT, cinfo.Sig = jwt, sig
+
+	} else if creds := c.leaf.remote.Credentials; creds != _EMPTY_ {
+		// Check for credentials first, that will take precedence..
 		c.Debugf("Authenticating with credentials file %q", c.leaf.remote.Credentials)
 		contents, err := ioutil.ReadFile(creds)
 		if err != nil {
diff --git a/server/monitor.go b/server/monitor.go
index 8c0b0b58..31824c0d 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1341,7 +1341,7 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 	<a href=.%s>JetStream</a>
 	<a href=.%s>Connections</a>
 	<a href=.%s>Accounts</a>
-	<a href=.%s>Account stats</a>
+	<a href=.%s>Account Stats</a>
 	<a href=.%s>Subscriptions</a>
 	<a href=.%s>Routes</a>
 	<a href=.%s>LeafNodes</a>
diff --git a/server/opts.go b/server/opts.go
index 0753464f..d45a3e20 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -159,18 +159,24 @@ type LeafNodeOpts struct {
 	tlsConfigOpts *TLSConfigOpts
 }
 
+// SignatureHandler is used to sign a nonce from the server while
+// authenticating with Nkeys. The callback should sign the nonce and
+// return the JWT and the raw signature.
+type SignatureHandler func([]byte) (string, []byte, error)
+
 // RemoteLeafOpts are options for connecting to a remote server as a leaf node.
 type RemoteLeafOpts struct {
-	LocalAccount string      `json:"local_account,omitempty"`
-	NoRandomize  bool        `json:"-"`
-	URLs         []*url.URL  `json:"urls,omitempty"`
-	Credentials  string      `json:"-"`
-	TLS          bool        `json:"-"`
-	TLSConfig    *tls.Config `json:"-"`
-	TLSTimeout   float64     `json:"tls_timeout,omitempty"`
-	Hub          bool        `json:"hub,omitempty"`
-	DenyImports  []string    `json:"-"`
-	DenyExports  []string    `json:"-"`
+	LocalAccount string           `json:"local_account,omitempty"`
+	NoRandomize  bool             `json:"-"`
+	URLs         []*url.URL       `json:"urls,omitempty"`
+	Credentials  string           `json:"-"`
+	SignatureCB  SignatureHandler `json:"-"`
+	TLS          bool             `json:"-"`
+	TLSConfig    *tls.Config      `json:"-"`
+	TLSTimeout   float64          `json:"tls_timeout,omitempty"`
+	Hub          bool             `json:"hub,omitempty"`
+	DenyImports  []string         `json:"-"`
+	DenyExports  []string         `json:"-"`
 
 	// When an URL has the "ws" (or "wss") scheme, then the server will initiate the
 	// connection as a websocket connection. By default, the websocket frames will be
diff --git a/server/route.go b/server/route.go
index aead8697..e8847118 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1608,7 +1608,7 @@ func (s *Server) updateRouteSubscriptionMap(acc *Account, sub *subscription, del
 		if ls, ok := lqws[key]; ok && ls == n {
 			acc.mu.Unlock()
 			return
-		} else {
+		} else if n > 0 {
 			lqws[key] = n
 		}
 		acc.mu.Unlock()
diff --git a/server/server.go b/server/server.go
index 02b812c3..3ac0b660 100644
--- a/server/server.go
+++ b/server/server.go
@@ -164,6 +164,7 @@ type Server struct {
 	leafRemoteCfgs     []*leafNodeCfg
 	leafRemoteAccounts sync.Map
 	leafNodeEnabled    bool
+	leafDisableConnect bool // Used in test only
 
 	quitCh           chan struct{}
 	startupComplete  chan struct{}
@@ -1625,6 +1626,11 @@ func (s *Server) Start() {
 	// Avoid RACE between Start() and Shutdown()
 	s.mu.Lock()
 	s.running = true
+	// Update leafNodeEnabled in case options have changed post NewServer()
+	// and before Start() (we should not be able to allow that, but server has
+	// direct reference to user-provided options - at least before a Reload() is
+	// performed.
+	s.leafNodeEnabled = opts.LeafNode.Port != 0 || len(opts.LeafNode.Remotes) > 0
 	s.mu.Unlock()
 
 	s.grMu.Lock()
diff --git a/server/stream.go b/server/stream.go
index e96f16db..9c023176 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1202,7 +1202,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 	}
 
 	// Here we will auto set direct gets if MaxMsgsPerSubject is set.
-	if cfg.MaxMsgsPer != 0 {
+	if cfg.MaxMsgsPer > 0 {
 		cfg.AllowDirect = true
 	}
 
@@ -4198,6 +4198,7 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	// Remove from our account map.
 	jsa.mu.Lock()
 	delete(jsa.streams, mset.cfg.Name)
+	accName := jsa.account.Name
 	jsa.mu.Unlock()
 
 	// Clean up consumers.
@@ -4318,6 +4319,12 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 			return err
 		}
 		js.releaseStreamResources(&mset.cfg)
+
+		// cleanup directories after the stream
+		accDir := filepath.Join(js.config.StoreDir, accName)
+		// no op if not empty
+		os.Remove(filepath.Join(accDir, streamsDir))
+		os.Remove(accDir)
 	} else if err := store.Stop(); err != nil {
 		return err
 	}


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index e28611ef..b6ff811c 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -23,6 +23,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"math/rand"
 	"os"
 	"path/filepath"
@@ -35,6 +36,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/jwt/v2"
+
 	"github.com/nats-io/nats.go"
 )
 
@@ -3680,6 +3683,309 @@ func TestJetStreamClusterPeerExclusionTag(t *testing.T) {
 	require_NoError(t, err)
 }
 
+func TestJetStreamClusterAccountPurge(t *testing.T) {
+	sysKp, syspub := createKey(t)
+	sysJwt := encodeClaim(t, jwt.NewAccountClaims(syspub), syspub)
+	sysCreds := newUser(t, sysKp)
+	defer removeFile(t, sysCreds)
+	accKp, accpub := createKey(t)
+	accClaim := jwt.NewAccountClaims(accpub)
+	accClaim.Limits.JetStreamLimits.DiskStorage = 1024 * 1024 * 5
+	accClaim.Limits.JetStreamLimits.MemoryStorage = 1024 * 1024 * 5
+	accJwt := encodeClaim(t, accClaim, accpub)
+	accCreds := newUser(t, accKp)
+	defer removeFile(t, accCreds)
+
+	tmlp := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		leaf {
+			listen: 127.0.0.1:-1
+		}
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+	`
+	c := createJetStreamClusterWithTemplateAndModHook(t, tmlp, "cluster", 3,
+		func(serverName, clustername, storeDir, conf string) string {
+			return conf + fmt.Sprintf(`
+				operator: %s
+				system_account: %s
+				resolver: {
+					type: full
+					dir: '%s/jwt'
+				}`, ojwt, syspub, storeDir)
+		})
+	defer c.shutdown()
+
+	c.waitOnLeader()
+
+	updateJwt(t, c.randomServer().ClientURL(), sysCreds, sysJwt, 3)
+	updateJwt(t, c.randomServer().ClientURL(), sysCreds, accJwt, 3)
+
+	createTestData := func() {
+		t.Helper()
+		nc := natsConnect(t, c.randomNonLeader().ClientURL(), nats.UserCredentials(accCreds))
+		defer nc.Close()
+
+		js, err := nc.JetStream()
+		require_NoError(t, err)
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:     "TEST1",
+			Subjects: []string{"foo"},
+			Replicas: 3,
+		})
+		require_NoError(t, err)
+
+		ci, err := js.AddConsumer("TEST1",
+			&nats.ConsumerConfig{Durable: "DUR1",
+				AckPolicy: nats.AckExplicitPolicy})
+		require_NoError(t, err)
+		require_True(t, ci.Config.Replicas == 0)
+
+		ci, err = js.AddConsumer("TEST1",
+			&nats.ConsumerConfig{Durable: "DUR2",
+				AckPolicy: nats.AckExplicitPolicy,
+				Replicas:  1})
+		require_NoError(t, err)
+		require_True(t, ci.Config.Replicas == 1)
+
+		toSend := uint64(1_000)
+		for i := uint64(0); i < toSend; i++ {
+			_, err = js.Publish("foo", nil)
+			require_NoError(t, err)
+		}
+
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:     "TEST2",
+			Subjects: []string{"bar"},
+			Replicas: 1,
+		})
+		require_NoError(t, err)
+
+		ci, err = js.AddConsumer("TEST2",
+			&nats.ConsumerConfig{Durable: "DUR1",
+				AckPolicy: nats.AckExplicitPolicy,
+				Replicas:  0})
+		require_NoError(t, err)
+		require_True(t, ci.Config.Replicas == 0)
+
+		for i := uint64(0); i < toSend; i++ {
+			_, err = js.Publish("bar", nil)
+			require_NoError(t, err)
+		}
+	}
+
+	inspectDirs := func(t *testing.T, sysTotal, accTotal int) error {
+		t.Helper()
+		sysDirs := 0
+		accDirs := 0
+		for _, s := range c.servers {
+			files, err := ioutil.ReadDir(s.getOpts().StoreDir + "/jetstream/" + syspub + "/_js_/")
+			require_NoError(t, err)
+			sysDirs += len(files) - 1 // sub 1 for _meta_
+			files, err = ioutil.ReadDir(s.getOpts().StoreDir + "/jetstream/" + accpub + "/streams/")
+			if err == nil || (err != nil && err.(*os.PathError).Error() == "no such file or directory") {
+				accDirs += len(files)
+			}
+		}
+		if sysDirs != sysTotal || accDirs != accTotal {
+			return fmt.Errorf("expected directory count does not match %d == %d, %d == %d",
+				sysDirs, sysTotal, accDirs, accTotal)
+		}
+		return nil
+	}
+
+	checkForDirs := func(t *testing.T, sysTotal, accTotal int) {
+		t.Helper()
+		checkFor(t, 20*time.Second, 250*time.Millisecond, func() error {
+			return inspectDirs(t, sysTotal, accTotal)
+		})
+	}
+
+	purge := func(t *testing.T) {
+		t.Helper()
+		ncsys, err := nats.Connect(c.randomServer().ClientURL(), nats.UserCredentials(sysCreds))
+		require_NoError(t, err)
+		defer ncsys.Close()
+
+		request := func() error {
+			var resp JSApiAccountPurgeResponse
+			m, err := ncsys.Request(fmt.Sprintf(JSApiAccountPurgeT, accpub), nil, time.Second)
+			if err != nil {
+				return err
+			}
+			if err := json.Unmarshal(m.Data, &resp); err != nil {
+				return err
+			}
+			if !resp.Initiated {
+				return fmt.Errorf("not started")
+			}
+			return nil
+		}
+		checkFor(t, 30*time.Second, 250*time.Millisecond, request)
+	}
+
+	t.Run("startup-cleanup", func(t *testing.T) {
+		_, newCleanupAcc1 := createKey(t)
+		_, newCleanupAcc2 := createKey(t)
+		for _, s := range c.servers {
+			os.MkdirAll(filepath.Join(s.getOpts().StoreDir, JetStreamStoreDir, newCleanupAcc1, streamsDir), defaultDirPerms)
+			os.MkdirAll(filepath.Join(s.getOpts().StoreDir, JetStreamStoreDir, newCleanupAcc2), defaultDirPerms)
+		}
+		createTestData()
+		checkForDirs(t, 6, 4)
+		c.stopAll()
+		c.restartAll()
+		for _, s := range c.servers {
+			accDir := filepath.Join(s.getOpts().StoreDir, JetStreamStoreDir, newCleanupAcc1)
+			_, e := os.Stat(filepath.Join(accDir, streamsDir))
+			require_Error(t, e)
+			require_True(t, os.IsNotExist(e))
+			_, e = os.Stat(accDir)
+			require_Error(t, e)
+			require_True(t, os.IsNotExist(e))
+			_, e = os.Stat(filepath.Join(s.getOpts().StoreDir, JetStreamStoreDir, newCleanupAcc2))
+			require_Error(t, e)
+			require_True(t, os.IsNotExist(e))
+		}
+		checkForDirs(t, 6, 4)
+	})
+
+	t.Run("purge-with-restart", func(t *testing.T) {
+		createTestData()
+		checkForDirs(t, 6, 4)
+		purge(t)
+		checkForDirs(t, 0, 0)
+		c.stopAll()
+		c.restartAll()
+		checkForDirs(t, 0, 0)
+	})
+
+	t.Run("purge-with-reuse", func(t *testing.T) {
+		createTestData()
+		checkForDirs(t, 6, 4)
+		purge(t)
+		checkForDirs(t, 0, 0)
+		createTestData()
+		checkForDirs(t, 6, 4)
+		purge(t)
+		checkForDirs(t, 0, 0)
+	})
+
+	t.Run("purge-deleted-account", func(t *testing.T) {
+		createTestData()
+		checkForDirs(t, 6, 4)
+		c.stopAll()
+		for _, s := range c.servers {
+			require_NoError(t, os.Remove(s.getOpts().StoreDir+"/jwt/"+accpub+".jwt"))
+		}
+		c.restartAll()
+		checkForDirs(t, 6, 4)
+		c.waitOnClusterReady() // unfortunately, this does not wait until leader is not catching up.
+		purge(t)
+		checkForDirs(t, 0, 0)
+		c.stopAll()
+		c.restartAll()
+		checkForDirs(t, 0, 0)
+	})
+}
+
+func TestJetStreamAccountPurge(t *testing.T) {
+	sysKp, syspub := createKey(t)
+	sysJwt := encodeClaim(t, jwt.NewAccountClaims(syspub), syspub)
+	sysCreds := newUser(t, sysKp)
+	defer removeFile(t, sysCreds)
+	accKp, accpub := createKey(t)
+	accClaim := jwt.NewAccountClaims(accpub)
+	accClaim.Limits.JetStreamLimits.DiskStorage = 1024 * 1024 * 5
+	accClaim.Limits.JetStreamLimits.MemoryStorage = 1024 * 1024 * 5
+	accJwt := encodeClaim(t, accClaim, accpub)
+	accCreds := newUser(t, accKp)
+	defer removeFile(t, accCreds)
+
+	storeDir := createDir(t, _EMPTY_)
+	defer os.RemoveAll(storeDir)
+
+	cfg := createConfFile(t, []byte(fmt.Sprintf(`
+        host: 127.0.0.1
+        port:-1
+        server_name: S1
+        operator: %s
+        system_account: %s
+        resolver: {
+                type: full
+                dir: '%s/jwt'
+        }
+        jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s/js'}
+`, ojwt, syspub, storeDir, storeDir)))
+	defer os.Remove(cfg)
+
+	s, o := RunServerWithConfig(cfg)
+	updateJwt(t, s.ClientURL(), sysCreds, sysJwt, 1)
+	updateJwt(t, s.ClientURL(), sysCreds, accJwt, 1)
+	defer s.Shutdown()
+
+	inspectDirs := func(t *testing.T, accTotal int) error {
+		t.Helper()
+		if accTotal == 0 {
+			files, err := ioutil.ReadDir(o.StoreDir + "/jetstream/" + accpub)
+			require_True(t, len(files) == accTotal || err != nil)
+		} else {
+			files, err := ioutil.ReadDir(o.StoreDir + "/jetstream/" + accpub + "/streams")
+			require_NoError(t, err)
+			require_True(t, len(files) == accTotal)
+		}
+		return nil
+	}
+
+	createTestData := func() {
+		nc := natsConnect(t, s.ClientURL(), nats.UserCredentials(accCreds))
+		defer nc.Close()
+		js, err := nc.JetStream()
+		require_NoError(t, err)
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:     "TEST1",
+			Subjects: []string{"foo"},
+		})
+		require_NoError(t, err)
+		_, err = js.AddConsumer("TEST1",
+			&nats.ConsumerConfig{Durable: "DUR1",
+				AckPolicy: nats.AckExplicitPolicy})
+		require_NoError(t, err)
+	}
+
+	purge := func(t *testing.T) {
+		t.Helper()
+		var resp JSApiAccountPurgeResponse
+		ncsys := natsConnect(t, s.ClientURL(), nats.UserCredentials(sysCreds))
+		defer ncsys.Close()
+		m, err := ncsys.Request(fmt.Sprintf(JSApiAccountPurgeT, accpub), nil, 5*time.Second)
+		require_NoError(t, err)
+		err = json.Unmarshal(m.Data, &resp)
+		require_NoError(t, err)
+		require_True(t, resp.Initiated)
+	}
+
+	createTestData()
+	inspectDirs(t, 1)
+	purge(t)
+	inspectDirs(t, 0)
+	createTestData()
+	inspectDirs(t, 1)
+
+	s.Shutdown()
+	require_NoError(t, os.Remove(storeDir+"/jwt/"+accpub+".jwt"))
+
+	s, o = RunServerWithConfig(o.ConfigFile)
+	inspectDirs(t, 1)
+	purge(t)
+	inspectDirs(t, 0)
+}
+
 func TestJetStreamClusterScaleConsumer(t *testing.T) {
 	c := createJetStreamClusterWithTemplate(t, jsClusterTempl, "C", 3)
 	defer c.shutdown()
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index a6325fc7..8c69e8c4 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -951,7 +951,7 @@ func (s *Server) closeAndDisableLeafnodes() {
 		leafs = append(leafs, ln)
 	}
 	// Disable leafnodes for now.
-	s.leafNodeEnabled = false
+	s.leafDisableConnect = true
 	s.mu.Unlock()
 
 	for _, ln := range leafs {
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index ac26e88f..22e229a5 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -4398,3 +4398,107 @@ func TestLeafNodeAuthConfigReload(t *testing.T) {
 	// Reload here should work ok.
 	reloadUpdateConfig(t, s, conf, template)
 }
+
+func TestLeafNodeSignatureCB(t *testing.T) {
+	content := `
+		port: -1
+		server_name: OP
+		operator = "../test/configs/nkeys/op.jwt"
+		resolver = MEMORY
+		listen: "127.0.0.1:-1"
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+	`
+	conf := createConfFile(t, []byte(content))
+	s, opts := RunServerWithConfig(conf)
+	defer removeFile(t, conf)
+	defer s.Shutdown()
+
+	_, akp := createAccount(s)
+	kp, _ := nkeys.CreateUser()
+	pub, _ := kp.PublicKey()
+	nuc := jwt.NewUserClaims(pub)
+	ujwt, err := nuc.Encode(akp)
+	if err != nil {
+		t.Fatalf("Error generating user JWT: %v", err)
+	}
+
+	lopts := &DefaultTestOptions
+	u, err := url.Parse(fmt.Sprintf("nats://%s:%d", opts.LeafNode.Host, opts.LeafNode.Port))
+	if err != nil {
+		t.Fatalf("Error parsing url: %v", err)
+	}
+	remote := &RemoteLeafOpts{URLs: []*url.URL{u}}
+	remote.SignatureCB = func(nonce []byte) (string, []byte, error) {
+		return "", nil, fmt.Errorf("on purpose")
+	}
+	lopts.LeafNode.Remotes = []*RemoteLeafOpts{remote}
+	lopts.LeafNode.ReconnectInterval = 100 * time.Millisecond
+	sl := RunServer(lopts)
+	defer sl.Shutdown()
+
+	slog := &captureErrorLogger{errCh: make(chan string, 10)}
+	sl.SetLogger(slog, false, false)
+
+	// Now check that the leafnode got the error that the callback returned.
+	select {
+	case err := <-slog.errCh:
+		if !strings.Contains(err, "on purpose") {
+			t.Fatalf("Expected error from cb, got %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Did not get expected error")
+	}
+
+	sl.Shutdown()
+	// Now check what happens if the connection is closed while in the callback.
+	blockCh := make(chan struct{})
+	remote.SignatureCB = func(nonce []byte) (string, []byte, error) {
+		<-blockCh
+		sig, err := kp.Sign(nonce)
+		return ujwt, sig, err
+	}
+	sl = RunServer(lopts)
+	defer sl.Shutdown()
+
+	// Recreate the logger so that we are sure not to have possible previous errors
+	slog = &captureErrorLogger{errCh: make(chan string, 10)}
+	sl.SetLogger(slog, false, false)
+
+	// Get the leaf connection from the temp clients map and close it.
+	checkFor(t, time.Second, 15*time.Millisecond, func() error {
+		var c *client
+		sl.grMu.Lock()
+		for _, cli := range sl.grTmpClients {
+			c = cli
+		}
+		sl.grMu.Unlock()
+		if c == nil {
+			return fmt.Errorf("Client still not found in temp map")
+		}
+		c.closeConnection(ClientClosed)
+		return nil
+	})
+
+	// Release the callback, and check we get the appropriate error.
+	close(blockCh)
+	select {
+	case err := <-slog.errCh:
+		if !strings.Contains(err, ErrConnectionClosed.Error()) {
+			t.Fatalf("Expected error that connection was closed, got %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Did not get expected error")
+	}
+
+	sl.Shutdown()
+	// Change to a good CB and now it should work
+	remote.SignatureCB = func(nonce []byte) (string, []byte, error) {
+		sig, err := kp.Sign(nonce)
+		return ujwt, sig, err
+	}
+	sl = RunServer(lopts)
+	defer sl.Shutdown()
+	checkLeafNodeConnected(t, sl)
+}
diff --git a/server/routes_test.go b/server/routes_test.go
index 69205ae7..0b552e9a 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -1526,3 +1526,67 @@ func TestSubjectRenameViaJetStreamAck(t *testing.T) {
 		t.Fatalf("Expected error")
 	}
 }
+
+func TestClusterQueueGroupWeightTrackingLeak(t *testing.T) {
+	o := DefaultOptions()
+	o.ServerName = "A"
+	s := RunServer(o)
+	defer s.Shutdown()
+
+	o2 := DefaultOptions()
+	o2.ServerName = "B"
+	o2.Routes = RoutesFromStr(fmt.Sprintf("nats://127.0.0.1:%d", o.Cluster.Port))
+	s2 := RunServer(o2)
+	defer s2.Shutdown()
+
+	nc := natsConnect(t, s.ClientURL())
+	defer nc.Close()
+
+	// Create a queue subscription
+	sub := natsQueueSubSync(t, nc, "foo", "bar")
+
+	// Check on s0 that we have the proper queue weight info
+	acc := s.GlobalAccount()
+
+	check := func(present bool, expected int32) {
+		t.Helper()
+		checkFor(t, time.Second, 15*time.Millisecond, func() error {
+			acc.mu.RLock()
+			v, ok := acc.lqws["foo bar"]
+			acc.mu.RUnlock()
+			if present {
+				if !ok {
+					return fmt.Errorf("the key is not present")
+				}
+				if v != expected {
+					return fmt.Errorf("lqws doest not contain expected value of %v: %v", expected, v)
+				}
+			} else if ok {
+				return fmt.Errorf("the key is present with value %v and should not be", v)
+			}
+			return nil
+		})
+	}
+	check(true, 1)
+
+	// Now unsub, and it should be removed, not just be 0
+	sub.Unsubscribe()
+	check(false, 0)
+
+	// Still make sure that the subject interest is gone from both servers.
+	checkSubGone := func(s *Server) {
+		t.Helper()
+		checkFor(t, time.Second, 15*time.Millisecond, func() error {
+			acc := s.GlobalAccount()
+			acc.mu.RLock()
+			res := acc.sl.Match("foo")
+			acc.mu.RUnlock()
+			if res != nil && len(res.qsubs) > 0 {
+				return fmt.Errorf("Found queue sub on foo for server %v", s)
+			}
+			return nil
+		})
+	}
+	checkSubGone(s)
+	checkSubGone(s2)
+}