Repository: nats-io/nats-server
Pull request #2927 (2022-03-17 03:57:47)

Improve consumer state handling.
Signed-off-by: Derek Collison <derek@nats.io>
 
/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 73bd2e42..e28f3836 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1760,7 +1760,7 @@ func (o *consumer) readStoredState() error {
 		return nil
 	}
 	state, err := o.store.State()
-	if err == nil && state != nil && state.Delivered.Consumer != 0 {
+	if err == nil && state != nil && (state.Delivered.Consumer != 0 || state.Delivered.Stream != 0) {
 		o.applyState(state)
 		if len(o.rdc) > 0 {
 			o.checkRedelivered()
@@ -3756,7 +3756,7 @@ func (o *consumer) setInitialPendingAndStart() {
 		}
 	}
 
-	if !filtered && dp != DeliverLastPerSubject {
+	if !filtered && (dp != DeliverLastPerSubject && dp != DeliverNew) {
 		var state StreamState
 		mset.store.FastState(&state)
 		if state.Msgs > 0 {
@@ -3777,7 +3777,12 @@ func (o *consumer) setInitialPendingAndStart() {
 			if dp == DeliverLast || dp == DeliverLastPerSubject {
 				o.sseq = ss.Last
 			} else if dp == DeliverNew {
-				o.sseq = ss.Last + 1
+				// If our original is larger we will ignore, we don't want to go backwards with DeliverNew.
+				// If its greater, we need to adjust pending.
+				if ss.Last >= o.sseq {
+					o.sgap -= (ss.Last - o.sseq + 1)
+					o.sseq = ss.Last + 1
+				}
 			} else {
 				// DeliverAll, DeliverByStartSequence, DeliverByStartTime
 				o.sseq = ss.First
@@ -3789,6 +3794,15 @@ func (o *consumer) setInitialPendingAndStart() {
 		}
 		o.updateSkipped()
 	}
+
+	// Update our persisted state if something has changed.
+	if store := o.store; store != nil {
+		if state, _ := store.State(); state != nil {
+			if o.dseq-1 > state.Delivered.Consumer || o.sseq-1 > state.Delivered.Stream {
+				o.writeStoreStateUnlocked()
+			}
+		}
+	}
 }
 
 func (o *consumer) decStreamPending(sseq uint64, subj string) {
diff --git a/server/filestore.go b/server/filestore.go
index d23cdf59..8d69b3af 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -5511,7 +5511,7 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 	state := &ConsumerState{}
 
 	// See if we have a running state or if we need to read in from disk.
-	if o.state.Delivered.Consumer != 0 {
+	if o.state.Delivered.Consumer != 0 || o.state.Delivered.Stream != 0 {
 		state.Delivered = o.state.Delivered
 		state.AckFloor = o.state.AckFloor
 		if len(o.state.Pending) > 0 {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 8d9afee1..eaf46302 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -3146,7 +3146,9 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			if recovering && !isLeader {
 				js.setConsumerAssignmentRecovering(ca)
 			}
-			js.processConsumerLeaderChange(o, isLeader)
+			if err := js.processConsumerLeaderChange(o, isLeader); err == nil && isLeader {
+				doSnapshot()
+			}
 		case <-t.C:
 			doSnapshot()
 		}
@@ -3320,10 +3322,17 @@ func decodeDeliveredUpdate(buf []byte) (dseq, sseq, dc uint64, ts int64, err err
 	return dseq, sseq, dc, ts, nil
 }
 
-func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) {
+func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) error {
+	stepDownIfLeader := func() error {
+		if node := o.raftNode(); node != nil && isLeader {
+			node.StepDown()
+		}
+		return errors.New("failed to update consumer leader status")
+	}
+
 	ca := o.consumerAssignment()
 	if ca == nil {
-		return
+		return stepDownIfLeader()
 	}
 	js.mu.Lock()
 	s, account, err := js.srv, ca.Client.serviceAccount(), ca.err
@@ -3336,7 +3345,7 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) {
 	consumerName := o.String()
 	acc, _ := s.LookupAccount(account)
 	if acc == nil {
-		return
+		return stepDownIfLeader()
 	}
 
 	if isLeader {
@@ -3370,7 +3379,7 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) {
 		if isLeader {
 			o.clearInitialInfo()
 		}
-		return
+		return nil
 	}
 
 	var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
@@ -3384,6 +3393,8 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) {
 			o.sendCreateAdvisory()
 		}
 	}
+
+	return nil
 }
 
 // Determines if we should send lost quorum advisory. We throttle these after first one.
diff --git a/server/monitor.go b/server/monitor.go
index 25b7c21c..036c6a4e 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2713,12 +2713,25 @@ func (s *Server) HandleHealthz(w http.ResponseWriter, r *http.Request) {
 	s.httpReqStats[HealthzPath]++
 	s.mu.Unlock()
 
+	hs := s.healthz()
+	if hs.Error != _EMPTY_ {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}
+	b, err := json.Marshal(hs)
+	if err != nil {
+		s.Errorf("Error marshaling response to /healthz request: %v", err)
+	}
+
+	ResponseHandler(w, r, b)
+}
+
+// Generate health status.
+func (s *Server) healthz() *HealthStatus {
 	var health = &HealthStatus{Status: "ok"}
 
 	if err := s.readyForConnections(time.Millisecond); err != nil {
 		health.Status = "error"
 		health.Error = err.Error()
-		w.WriteHeader(http.StatusServiceUnavailable)
 	} else if js := s.getJetStream(); js != nil {
 		// Check JetStream status here.
 		js.mu.RLock()
@@ -2731,11 +2744,9 @@ func (s *Server) HandleHealthz(w http.ResponseWriter, r *http.Request) {
 			if node.GroupLeader() == _EMPTY_ {
 				health.Status = "unavailable"
 				health.Error = "JetStream has not established contact with a meta leader"
-				w.WriteHeader(http.StatusServiceUnavailable)
 			} else if !node.Current() {
 				health.Status = "unavailable"
 				health.Error = "JetStream is not current with the meta leader"
-				w.WriteHeader(http.StatusServiceUnavailable)
 			} else {
 				// If we are here we are current and have seen our meta leader.
 				// Now check assets.
@@ -2772,21 +2783,23 @@ func (s *Server) HandleHealthz(w http.ResponseWriter, r *http.Request) {
 						if !cc.isStreamCurrent(accName, sname) {
 							health.Status = "unavailable"
 							health.Error = fmt.Sprintf("JetStream stream %q for account %q is not current", sname, accName)
-							w.WriteHeader(http.StatusServiceUnavailable)
 							js.mu.RUnlock()
 							break Err
 						}
+						// Now do consumers.
+						for _, o := range stream.getConsumers() {
+							if node := o.raftNode(); node != nil && !node.Current() {
+								health.Status = "unavailable"
+								health.Error = fmt.Sprintf("JetStream consumer %q for stream %q and account %q is not current", o.String(), sname, accName)
+								js.mu.RUnlock()
+								break Err
+							}
+						}
 					}
 					js.mu.RUnlock()
 				}
 			}
 		}
 	}
-
-	b, err := json.Marshal(health)
-	if err != nil {
-		s.Errorf("Error marshaling response to /healthz request: %v", err)
-	}
-
-	ResponseHandler(w, r, b)
+	return health
 }


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index cb5db165..0f89c448 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -11206,6 +11206,71 @@ func TestJetStreamRemovedPeersAndStreamsListAndDelete(t *testing.T) {
 	require_Error(t, err, nats.ErrStreamNotFound)
 }
 
+func TestJetStreamConsumerDeliverNewBug(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	pcn := "C2"
+	sc.waitOnLeader()
+	ml := sc.leader()
+	if ml.ClusterName() == pcn {
+		pcn = "C1"
+	}
+
+	// Client based API
+	nc, js := jsClientConnect(t, ml)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "T",
+		Replicas:  3,
+		Placement: &nats.Placement{Cluster: pcn},
+	})
+	require_NoError(t, err)
+
+	// Put messages in..
+	num := 200
+	for i := 0; i < num; i++ {
+		js.PublishAsync("T", []byte("OK"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	ci, err := js.AddConsumer("T", &nats.ConsumerConfig{
+		Durable:       "d",
+		AckPolicy:     nats.AckExplicitPolicy,
+		DeliverPolicy: nats.DeliverNewPolicy,
+	})
+	require_NoError(t, err)
+
+	if ci.Delivered.Consumer != 0 || ci.Delivered.Stream != 200 {
+		t.Fatalf("Incorrect consumer delivered info: %+v", ci.Delivered)
+	}
+
+	c := sc.clusterForName(pcn)
+	for _, s := range c.servers {
+		sd := s.JetStreamConfig().StoreDir
+		s.Shutdown()
+		removeDir(t, sd)
+		s = c.restartServer(s)
+		c.waitOnServerHealthz(s)
+	}
+
+	c.waitOnConsumerLeader("$G", "T", "d")
+	ci, err = js.ConsumerInfo("T", "d")
+	require_NoError(t, err)
+
+	if ci.Delivered.Consumer != 0 || ci.Delivered.Stream != 200 {
+		t.Fatalf("Incorrect consumer delivered info: %+v", ci.Delivered)
+	}
+	if ci.NumPending != 0 {
+		t.Fatalf("Did not expect NumPending, got %d", ci.NumPending)
+	}
+}
+
 // Support functions
 
 // Used to setup superclusters for tests.
@@ -12143,6 +12208,19 @@ func (c *cluster) waitOnStreamCurrent(s *Server, account, stream string) {
 	c.t.Fatalf("Expected server %q to eventually be current for stream %q", s, stream)
 }
 
+func (c *cluster) waitOnServerHealthz(s *Server) {
+	c.t.Helper()
+	expires := time.Now().Add(30 * time.Second)
+	for time.Now().Before(expires) {
+		hs := s.healthz()
+		if hs.Status == "ok" && hs.Error == _EMPTY_ {
+			return
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	c.t.Fatalf("Expected server %q to eventually return healthz 'ok', but got %q", s, s.healthz().Error)
+}
+
 func (c *cluster) waitOnServerCurrent(s *Server) {
 	c.t.Helper()
 	expires := time.Now().Add(20 * time.Second)