[FIXED] Ghost consumers after failed meta proposal (#6088)

If the `cc.meta.Propose` fails, for example due to restarts, the
`sa.consumers[ca.Name] = ca` would have staged a consumer that will
never be successfully added. That consumer would then become part of the
next snapshot (for example during shutdown), which meant that that
server had a ghost consumer it was unable to clean up.

This does not fully solve the ghost consumers issue, but this should
contribute to the final solution.

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

server/jetstream_cluster.go

@@ -138,10 +138,11 @@ type streamAssignment struct {
 	Reply   string        `json:"reply"`
 	Restore *StreamState  `json:"restore_state,omitempty"`
 	// Internal
-	consumers  map[string]*consumerAssignment
-	responded  bool
-	recovering bool
-	err        error
+	consumers        map[string]*consumerAssignment
+	pendingConsumers map[string]struct{}
+	responded        bool
+	recovering       bool
+	err              error
 }
 
 // consumerAssignment is what the meta controller uses to assign consumers to streams.
@@ -4264,6 +4265,10 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	// Place into our internal map under the stream assignment.
 	// Ok to replace an existing one, we check on process call below.
 	sa.consumers[ca.Name] = ca
+	delete(sa.pendingConsumers, ca.Name)
+	if len(sa.pendingConsumers) == 0 {
+		sa.pendingConsumers = nil
+	}
 	js.mu.Unlock()
 
 	acc, err := s.LookupAccount(accName)
@@ -7415,7 +7420,7 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		}
 		if maxc > 0 {
 			// Don't count DIRECTS.
-			total := 0
+			total := len(sa.pendingConsumers)
 			for cn, ca := range sa.consumers {
 				if action == ActionCreateOrUpdate {
 					// If the consumer name is specified and we think it already exists, then
@@ -7673,14 +7678,14 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		ca = nca
 	}
 
-	// Mark this as pending.
-	if sa.consumers == nil {
-		sa.consumers = make(map[string]*consumerAssignment)
-	}
-	sa.consumers[ca.Name] = ca
-
 	// Do formal proposal.
-	cc.meta.Propose(encodeAddConsumerAssignment(ca))
+	if err := cc.meta.Propose(encodeAddConsumerAssignment(ca)); err == nil {
+		// Mark this as pending.
+		if sa.pendingConsumers == nil {
+			sa.pendingConsumers = make(map[string]struct{})
+		}
+		sa.pendingConsumers[ca.Name] = struct{}{}
+	}
 }
 
 func encodeAddConsumerAssignment(ca *consumerAssignment) []byte {

server/jetstream_cluster_2_test.go

@@ -2072,6 +2072,12 @@ func TestJetStreamClusterMaxConsumersMultipleConcurrentRequests(t *testing.T) {
 	if nc := len(names); nc > 1 {
 		t.Fatalf("Expected only 1 consumer, got %d", nc)
 	}
+
+	metaLeader := c.leader()
+	mjs := metaLeader.getJetStream()
+	sa := mjs.streamAssignment(globalAccountName, "MAXCC")
+	require_NotNil(t, sa)
+	require_True(t, sa.pendingConsumers == nil)
 }
 
 func TestJetStreamClusterAccountMaxStreamsAndConsumersMultipleConcurrentRequests(t *testing.T) {