Fix desync after errCatchupTooManyRetries

server/jetstream_cluster.go

@@ -2845,9 +2845,14 @@ func (mset *stream) resetClusteredState(err error) bool {
 		return false
 	}
 
-	// We delete our raft state. Will recreate.
 	if node != nil {
-		node.Delete()
+		if err == errCatchupTooManyRetries {
+			// Don't delete all state, could've just been temporarily unable to reach the leader.
+			node.Stop()
+		} else {
+			// We delete our raft state. Will recreate.
+			node.Delete()
+		}
 	}
 
 	// Preserve our current state and messages unless we have a first sequence mismatch.
@@ -8383,7 +8388,7 @@ RETRY:
 	}
 
 	numRetries++
-	if numRetries >= maxRetries {
+	if numRetries > maxRetries {
 		// Force a hard reset here.
 		return errCatchupTooManyRetries
 	}

server/jetstream_cluster_4_test.go

@@ -4086,3 +4086,76 @@ func TestJetStreamConsumerReplicasAfterScale(t *testing.T) {
 	require_Equal(t, ci.Config.Replicas, 3)
 	require_Equal(t, len(ci.Cluster.Replicas), 2)
 }
+
+func TestJetStreamClusterDesyncAfterCatchupTooManyRetries(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	si, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	streamLeader := si.Cluster.Leader
+	streamLeaderServer := c.serverByName(streamLeader)
+	nc.Close()
+	nc, js = jsClientConnect(t, streamLeaderServer)
+	defer nc.Close()
+
+	servers := slices.DeleteFunc([]string{"S-1", "S-2", "S-3"}, func(s string) bool {
+		return s == streamLeader
+	})
+
+	// Publish 10 messages.
+	for i := 0; i < 10; i++ {
+		pubAck, err := js.Publish("foo", []byte("ok"))
+		require_NoError(t, err)
+		require_Equal(t, pubAck.Sequence, uint64(i+1))
+	}
+
+	outdatedServerName := servers[0]
+	clusterResetServerName := servers[1]
+
+	outdatedServer := c.serverByName(outdatedServerName)
+	outdatedServer.Shutdown()
+	outdatedServer.WaitForShutdown()
+
+	// Publish 10 more messages, one server will be behind.
+	for i := 0; i < 10; i++ {
+		pubAck, err := js.Publish("foo", []byte("ok"))
+		require_NoError(t, err)
+		require_Equal(t, pubAck.Sequence, uint64(i+11))
+	}
+
+	// We will not need the client anymore.
+	nc.Close()
+
+	// Shutdown stream leader so one server remains.
+	streamLeaderServer.Shutdown()
+	streamLeaderServer.WaitForShutdown()
+
+	clusterResetServer := c.serverByName(clusterResetServerName)
+	acc, err := clusterResetServer.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err := acc.lookupStream("TEST")
+	require_NoError(t, err)
+
+	// Too many retries while processing snapshot is considered a cluster reset.
+	// If a leader is temporarily unavailable we shouldn't blow away our state.
+	require_True(t, isClusterResetErr(errCatchupTooManyRetries))
+	mset.resetClusteredState(errCatchupTooManyRetries)
+
+	// Stream leader stays offline, we only start the server with missing stream data.
+	// We expect that the reset server must not allow the outdated server to become leader, as that would result in desync.
+	c.restartServer(outdatedServer)
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	// Outdated server must NOT become the leader.
+	newStreamLeaderServer := c.streamLeader(globalAccountName, "TEST")
+	require_Equal(t, newStreamLeaderServer.Name(), clusterResetServerName)
+}