Cherry-picks for 2.10.17 (#5598)

Includes:

* #5593
* #5597
* #5600

Signed-off-by: Neil Twigg <[email protected]>

Author: wallyqs

server/client.go

@@ -847,7 +847,7 @@ func (c *client) applyAccountLimits() {
 	c.msubs = jwt.NoLimit
 	if c.opts.JWT != _EMPTY_ { // user jwt implies account
 		if uc, _ := jwt.DecodeUserClaims(c.opts.JWT); uc != nil {
-			c.mpay = int32(uc.Limits.Payload)
+			atomic.StoreInt32(&c.mpay, int32(uc.Limits.Payload))
 			c.msubs = int32(uc.Limits.Subs)
 			if uc.IssuerAccount != _EMPTY_ && uc.IssuerAccount != uc.Issuer {
 				if scope, ok := c.acc.signingKeys[uc.Issuer]; ok {
@@ -5427,7 +5427,7 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) {
 
 		if genid := atomic.LoadUint64(&sl.genid); genid != pac.genid {
 			ok = false
-			delete(c.in.pacache, bytesToString(c.pa.pacache))
+			c.in.pacache = make(map[string]*perAccountCache)
 		} else {
 			acc = pac.acc
 			r = pac.results

server/filestore.go

@@ -7635,10 +7635,23 @@ func (fs *fileStore) Delete() error {
 		os.RemoveAll(pdir)
 	}
 
-	// Do Purge() since if we have lots of blocks uses a mv/rename.
-	fs.Purge()
+	// Quickly close all blocks and simulate a purge w/o overhead an new write block.
+	fs.mu.Lock()
+	for _, mb := range fs.blks {
+		mb.dirtyClose()
+	}
+	dmsgs := fs.state.Msgs
+	dbytes := int64(fs.state.Bytes)
+	fs.state.Msgs, fs.state.Bytes = 0, 0
+	fs.blks = nil
+	cb := fs.scb
+	fs.mu.Unlock()
+
+	if cb != nil {
+		cb(-int64(dmsgs), -dbytes, 0, _EMPTY_)
+	}
 
-	if err := fs.stop(false); err != nil {
+	if err := fs.stop(true, false); err != nil {
 		return err
 	}
 
@@ -7654,14 +7667,19 @@ func (fs *fileStore) Delete() error {
 	// Do this in separate Go routine in case lots of blocks.
 	// Purge above protects us as does the removal of meta artifacts above.
 	go func() {
+		<-dios
 		err := os.RemoveAll(ndir)
+		dios <- struct{}{}
 		if err == nil {
 			return
 		}
 		ttl := time.Now().Add(time.Second)
 		for time.Now().Before(ttl) {
 			time.Sleep(10 * time.Millisecond)
-			if err = os.RemoveAll(ndir); err == nil {
+			<-dios
+			err = os.RemoveAll(ndir)
+			dios <- struct{}{}
+			if err == nil {
 				return
 			}
 		}
@@ -7927,11 +7945,11 @@ func (fs *fileStore) _writeFullState(force bool) error {
 
 // Stop the current filestore.
 func (fs *fileStore) Stop() error {
-	return fs.stop(true)
+	return fs.stop(false, true)
 }
 
 // Stop the current filestore.
-func (fs *fileStore) stop(writeState bool) error {
+func (fs *fileStore) stop(delete, writeState bool) error {
 	fs.mu.Lock()
 	if fs.closed || fs.closing {
 		fs.mu.Unlock()
@@ -7982,7 +8000,11 @@ func (fs *fileStore) stop(writeState bool) error {
 	fs.cmu.Unlock()
 
 	for _, o := range cfs {
-		o.Stop()
+		if delete {
+			o.StreamDelete()
+		} else {
+			o.Stop()
+		}
 	}
 
 	if bytes > 0 && cb != nil {

server/jetstream_cluster.go

@@ -4109,7 +4109,7 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		sa.consumers = make(map[string]*consumerAssignment)
 	} else if oca := sa.consumers[ca.Name]; oca != nil {
 		wasExisting = true
-		// Copy over private existing state from former SA.
+		// Copy over private existing state from former CA.
 		if ca.Group != nil {
 			ca.Group.node = oca.Group.node
 		}
@@ -4231,6 +4231,7 @@ func (js *jetStream) processConsumerRemoval(ca *consumerAssignment) {
 			if ca.Group != nil && oca.Group != nil && ca.Group.Name == oca.Group.Name {
 				needDelete = true
 				oca.deleted = true
+				oca.Group.node = nil
 				delete(sa.consumers, ca.Name)
 			}
 		}
@@ -7760,7 +7761,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		if err == nil {
 			err = NewJSAccountResourcesExceededError()
 		}
-		s.RateLimitWarnf(err.Error())
+		s.RateLimitWarnf("JetStream account limits exceeded for '%s': %s", jsa.acc().GetName(), err.Error())
 		if canRespond {
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = err

server/raft.go

@@ -275,6 +275,7 @@ var (
 	errLeaderLen         = fmt.Errorf("raft: leader should be exactly %d bytes", idLen)
 	errTooManyEntries    = errors.New("raft: append entry can contain a max of 64k entries")
 	errBadAppendEntry    = errors.New("raft: append entry corrupt")
+	errNoInternalClient  = errors.New("raft: no internal client")
 )
 
 // This will bootstrap a raftNode by writing its config into the store directory.
@@ -1625,6 +1626,13 @@ func (n *raft) shutdown(shouldDelete bool) {
 	// allowing shutdown() to be called again. If that happens then the below
 	// close(n.quit) will panic from trying to close an already-closed channel.
 	if n.state.Swap(int32(Closed)) == int32(Closed) {
+		// If we get called again with shouldDelete, in case we were called first with Stop() cleanup
+		if shouldDelete {
+			if wal := n.wal; wal != nil {
+				wal.Delete()
+			}
+			os.RemoveAll(n.sd)
+		}
 		n.Unlock()
 		return
 	}
@@ -1641,17 +1649,22 @@ func (n *raft) shutdown(shouldDelete bool) {
 			n.unsubscribe(sub)
 		}
 		c.closeConnection(InternalClient)
+		n.c = nil
 	}
+
 	s, g, wal := n.s, n.group, n.wal
 
 	// Unregistering ipQueues do not prevent them from push/pop
 	// just will remove them from the central monitoring map
 	queues := []interface {
 		unregister()
+		drain()
 	}{n.reqs, n.votes, n.prop, n.entry, n.resp, n.apply, n.stepdown}
 	for _, q := range queues {
+		q.drain()
 		q.unregister()
 	}
+	sd := n.sd
 	n.Unlock()
 
 	s.unregisterRaftNode(g)
@@ -1666,7 +1679,7 @@ func (n *raft) shutdown(shouldDelete bool) {
 
 	if shouldDelete {
 		// Delete all our peer state and vote state and any snapshots.
-		os.RemoveAll(n.sd)
+		os.RemoveAll(sd)
 		n.debug("Deleted")
 	} else {
 		n.debug("Shutdown")
@@ -1721,12 +1734,15 @@ func (n *raft) newInbox() string {
 // Our internal subscribe.
 // Lock should be held.
 func (n *raft) subscribe(subject string, cb msgHandler) (*subscription, error) {
+	if n.c == nil {
+		return nil, errNoInternalClient
+	}
 	return n.s.systemSubscribe(subject, _EMPTY_, false, n.c, cb)
 }
 
 // Lock should be held.
 func (n *raft) unsubscribe(sub *subscription) {
-	if sub != nil {
+	if n.c != nil && sub != nil {
 		n.c.processUnsub(sub.sid)
 	}
 }