[IMPROVED] When we detect a bump in clfs try to snapshot (if prior was zero force).

server/jetstream_cluster.go

@@ -2425,6 +2425,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			return
 		case <-aq.ch:
 			var ne, nb uint64
+			// If we bump clfs we will want to write out snapshot if within our time window.
+			pclfs := mset.getCLFS()
+
 			ces := aq.pop()
 			for _, ce := range ces {
 				// No special processing needed for when we are caught up on restart.
@@ -2441,6 +2444,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 					}
 					continue
 				}
+
 				// Apply our entries.
 				if err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
 					// Update our applied.
@@ -2472,7 +2476,13 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 
 			// Check about snapshotting
 			// If we have at least min entries to compact, go ahead and try to snapshot/compact.
-			if ne >= compactNumMin || nb > compactSizeMin {
+			if ne >= compactNumMin || nb > compactSizeMin || mset.getCLFS() > pclfs {
+				// We want to make sure we do not short circuit if transistioning from no clfs.
+				if pclfs == 0 {
+					// This is always false by default.
+					lastState.firstNeedsUpdate = true
+					lastSnapTime = time.Time{}
+				}
 				doSnapshot()
 			}
 

server/jetstream_cluster_4_test.go

@@ -2549,3 +2549,130 @@ func TestJetStreamClusterMetaSyncOrphanCleanup(t *testing.T) {
 		require_Equal(t, state.Msgs, 10)
 	}
 }
+
+func TestJetStreamClusterKeyValueDesyncAfterHardKill(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3F", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.serverByName("S-1"))
+	defer nc.Close()
+
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
+		Bucket:   "inconsistency",
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	// First create should succeed.
+	revision, err := kv.Create("key.exists", []byte("1"))
+	require_NoError(t, err)
+	require_Equal(t, revision, 1)
+
+	// Second create will be rejected but bump CLFS.
+	_, err = kv.Create("key.exists", []byte("2"))
+	require_Error(t, err)
+
+	// Insert a new message, should only be applied once, even if we hard kill and replay afterward.
+	revision, err = kv.Put("key.put", []byte("3"))
+	require_NoError(t, err)
+	require_Equal(t, revision, 2)
+
+	// Restart a server
+	s3 := c.serverByName("S-3")
+	// We will remove the index.db file after we shutdown.
+	mset, err := s3.GlobalAccount().lookupStream("KV_inconsistency")
+	require_NoError(t, err)
+	fs := mset.store.(*fileStore)
+	ifile := filepath.Join(fs.fcfg.StoreDir, msgDir, "index.db")
+
+	s3.Shutdown()
+	s3.WaitForShutdown()
+	// Remove the index.db file to simulate a hard kill where server can not write out the index.db file.
+	require_NoError(t, os.Remove(ifile))
+
+	c.restartServer(s3)
+	c.waitOnClusterReady()
+	c.waitOnAllCurrent()
+
+	getStreamDetails := func(t *testing.T, c *cluster, accountName, streamName string) *StreamDetail {
+		t.Helper()
+		srv := c.streamLeader(accountName, streamName)
+		if srv == nil {
+			return nil
+		}
+		jsz, err := srv.Jsz(&JSzOptions{Accounts: true, Streams: true, Consumer: true})
+		require_NoError(t, err)
+		for _, acc := range jsz.AccountDetails {
+			if acc.Name == accountName {
+				for _, stream := range acc.Streams {
+					if stream.Name == streamName {
+						return &stream
+					}
+				}
+			}
+		}
+		t.Error("Could not find account details")
+		return nil
+	}
+
+	checkState := func(t *testing.T, c *cluster, accountName, streamName string) error {
+		t.Helper()
+
+		leaderSrv := c.streamLeader(accountName, streamName)
+		if leaderSrv == nil {
+			return fmt.Errorf("no leader server found for stream %q", streamName)
+		}
+		streamLeader := getStreamDetails(t, c, accountName, streamName)
+		if streamLeader == nil {
+			return fmt.Errorf("no leader found for stream %q", streamName)
+		}
+		var errs []error
+		for _, srv := range c.servers {
+			if srv == leaderSrv {
+				// Skip self
+				continue
+			}
+			acc, err := srv.LookupAccount(accountName)
+			require_NoError(t, err)
+			stream, err := acc.lookupStream(streamName)
+			require_NoError(t, err)
+			state := stream.state()
+
+			if state.Msgs != streamLeader.State.Msgs {
+				err := fmt.Errorf("[%s] Leader %v has %d messages, Follower %v has %d messages",
+					streamName, leaderSrv, streamLeader.State.Msgs,
+					srv, state.Msgs,
+				)
+				errs = append(errs, err)
+			}
+			if state.FirstSeq != streamLeader.State.FirstSeq {
+				err := fmt.Errorf("[%s] Leader %v FirstSeq is %d, Follower %v is at %d",
+					streamName, leaderSrv, streamLeader.State.FirstSeq,
+					srv, state.FirstSeq,
+				)
+				errs = append(errs, err)
+			}
+			if state.LastSeq != streamLeader.State.LastSeq {
+				err := fmt.Errorf("[%s] Leader %v LastSeq is %d, Follower %v is at %d",
+					streamName, leaderSrv, streamLeader.State.LastSeq,
+					srv, state.LastSeq,
+				)
+				errs = append(errs, err)
+			}
+			if state.NumDeleted != streamLeader.State.NumDeleted {
+				err := fmt.Errorf("[%s] Leader %v NumDeleted is %d, Follower %v is at %d",
+					streamName, leaderSrv, streamLeader.State.NumDeleted,
+					srv, state.NumDeleted,
+				)
+				errs = append(errs, err)
+			}
+		}
+		if len(errs) > 0 {
+			return errors.Join(errs...)
+		}
+		return nil
+	}
+
+	err = checkState(t, c, "$G", "KV_inconsistency")
+	require_NoError(t, err)
+}

server/stream.go

@@ -1024,6 +1024,9 @@ func (mset *stream) lastSeqAndCLFS() (uint64, uint64) {
 }
 
 func (mset *stream) getCLFS() uint64 {
+	if mset == nil {
+		return 0
+	}
 	mset.clMu.Lock()
 	defer mset.clMu.Unlock()
 	return mset.clfs