Warn of consumer state update failures.

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

server/consumer.go

@@ -2100,6 +2100,7 @@ func (o *consumer) ackWait(next time.Duration) time.Duration {
 }
 
 // Due to bug in calculation of sequences on restoring redelivered let's do quick sanity check.
+// Lock should be held.
 func (o *consumer) checkRedelivered(slseq uint64) {
 	var lseq uint64
 	if mset := o.mset; mset != nil {
@@ -2114,7 +2115,13 @@ func (o *consumer) checkRedelivered(slseq uint64) {
 		}
 	}
 	if shouldUpdateState {
-		o.writeStoreStateUnlocked()
+		if err := o.writeStoreStateUnlocked(); err != nil && o.srv != nil && o.mset != nil {
+			s, acc, mset, name := o.srv, o.acc, o.mset, o.name
+			// Can not hold lock while gather information about account and stream below.
+			o.mu.Unlock()
+			s.Warnf("Consumer '%s > %s > %s' error on write store state from check redelivered: %v", acc, mset.name(), name, err)
+			o.mu.Lock()
+		}
 	}
 }
 
@@ -3882,7 +3889,13 @@ func (o *consumer) checkPending() {
 
 	// Update our state if needed.
 	if shouldUpdateState {
-		o.writeStoreStateUnlocked()
+		if err := o.writeStoreStateUnlocked(); err != nil && o.srv != nil && o.mset != nil {
+			s, acc, mset, name := o.srv, o.acc, o.mset, o.name
+			// Can not hold lock while gather information about account and stream below.
+			o.mu.Unlock()
+			s.Warnf("Consumer '%s > %s > %s' error on write store state from check pending: %v", acc, mset.name(), name, err)
+			o.mu.Lock()
+		}
 	}
 }
 
@@ -4169,9 +4182,13 @@ func (o *consumer) purge(sseq uint64, slseq uint64) {
 			}
 		}
 	}
+	// Grab some info in case of error below.
+	s, acc, mset, name := o.srv, o.acc, o.mset, o.name
 	o.mu.Unlock()
 
-	o.writeStoreState()
+	if err := o.writeStoreState(); err != nil && s != nil && mset != nil {
+		s.Warnf("Consumer '%s > %s > %s' error on write store state from purge: %v", acc, mset.name(), name, err)
+	}
 }
 
 func stopAndClearTimer(tp **time.Timer) {

server/jetstream_cluster.go

@@ -4206,7 +4206,14 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 				}
 				panic(err.Error())
 			}
-			o.store.Update(state)
+			if err = o.store.Update(state); err != nil {
+				o.mu.RLock()
+				s, acc, mset, name := o.srv, o.acc, o.mset, o.name
+				o.mu.RUnlock()
+				if s != nil && mset != nil {
+					s.Warnf("Consumer '%s > %s > %s' error on store update from snapshot entry: %v", acc, mset.name(), name, err)
+				}
+			}
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
 			var ourID string

server/jetstream_test.go

@@ -19558,8 +19558,7 @@ func TestJetStreamPartialPurgeWithAckPending(t *testing.T) {
 
 	nmsgs := 100
 	for i := 0; i < nmsgs; i++ {
-		_, err := js.Publish("foo", []byte("OK"))
-		require_NoError(t, err)
+		sendStreamMsg(t, nc, "foo", "OK")
 	}
 	sub, err := js.PullSubscribe("foo", "dlc", nats.AckWait(time.Second))
 	require_NoError(t, err)
@@ -19583,8 +19582,7 @@ func TestJetStreamPartialPurgeWithAckPending(t *testing.T) {
 	require_True(t, ci.NumPending == 0)
 
 	for i := 0; i < nmsgs; i++ {
-		_, err := js.Publish("foo", []byte("OK"))
-		require_NoError(t, err)
+		sendStreamMsg(t, nc, "foo", "OK")
 	}
 
 	ci, err = js.ConsumerInfo("TEST", "dlc")
@@ -19666,8 +19664,7 @@ func TestJetStreamPurgeWithRedeliveredPending(t *testing.T) {
 
 	nmsgs := 100
 	for i := 0; i < nmsgs; i++ {
-		_, err := js.Publish("foo", []byte("OK"))
-		require_NoError(t, err)
+		sendStreamMsg(t, nc, "foo", "OK")
 	}
 	sub, err := js.PullSubscribe("foo", "dlc", nats.AckWait(time.Second))
 	require_NoError(t, err)
@@ -19715,8 +19712,7 @@ func TestJetStreamConsumerAckFloorWithExpired(t *testing.T) {
 
 	nmsgs := 100
 	for i := 0; i < nmsgs; i++ {
-		_, err := js.Publish("foo", []byte("OK"))
-		require_NoError(t, err)
+		sendStreamMsg(t, nc, "foo", "OK")
 	}
 	sub, err := js.PullSubscribe("foo", "dlc", nats.AckWait(time.Second))
 	require_NoError(t, err)