[IMPROVED] Check consumer leader status without locks. (#5386)

Saw TestJetStreamClusterWorkQueueAfterScaleUp locking up, so adjusted
the leader check to the atomic bool.

Signed-off-by: Derek Collison <derek@nats.io>

server/consumer.go

@@ -5394,9 +5394,14 @@ func (o *consumer) signalSubs() []*subscription {
 // but we must check if we are leader.
 // We do need the sequence of the message however and we use the msg as the encoded seq.
 func (o *consumer) processStreamSignal(_ *subscription, _ *client, _ *Account, subject, _ string, seqb []byte) {
+	// We can get called here now when not leader, so bail fast
+	// and without acquiring any locks.
+	if !o.leader.Load() {
+		return
+	}
 	o.mu.Lock()
 	defer o.mu.Unlock()
-	if o.mset == nil || !o.isLeader() {
+	if o.mset == nil {
 		return
 	}
 

server/jetstream_cluster_3_test.go

@@ -3067,7 +3067,12 @@ func TestJetStreamClusterWorkQueueAfterScaleUp(t *testing.T) {
 	c.waitOnStreamLeader(globalAccountName, "TEST")
 
 	sendStreamMsg(t, nc, "WQ", "SOME WORK")
-	<-wch
+
+	select {
+	case <-wch:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive ack signal")
+	}
 
 	checkFor(t, time.Second, 200*time.Millisecond, func() error {
 		si, err := js.StreamInfo("TEST")