func (c *cluster) SetVersion(ver *semver.Version) {
	c.Lock()
	defer c.Unlock()
	if c.version != nil {
		plog.Noticef("updated the cluster version from %v to %v", version.Cluster(c.version.String()), version.Cluster(ver.String()))
	} else {
		plog.Noticef("set the initial cluster version to %v", version.Cluster(ver.String()))
	}
	c.version = ver
	MustDetectDowngrade(c.version)
}

func (c *RaftCluster) SetVersion(ver *semver.Version) {
	c.Lock()
	defer c.Unlock()
	if c.version != nil {
		plog.Noticef("updated the cluster version from %v to %v", version.Cluster(c.version.String()), version.Cluster(ver.String()))
	} else {
		plog.Noticef("set the initial cluster version to %v", version.Cluster(ver.String()))
	}
	c.version = ver
	mustDetectDowngrade(c.version)
	if c.store != nil {
		mustSaveClusterVersionToStore(c.store, ver)
	}
	if c.be != nil {
		mustSaveClusterVersionToBackend(c.be, ver)
	}
}

func (c *cluster) Recover() {
	c.members, c.removed = membersFromStore(c.store)
	c.version = clusterVersionFromStore(c.store)
	MustDetectDowngrade(c.version)

	for _, m := range c.members {
		plog.Infof("added member %s %v to cluster %s from store", m.ID, m.PeerURLs, c.id)
	}
	if c.version != nil {
		plog.Infof("set the cluster version to %v from store", version.Cluster(c.version.String()))
	}
}

func (s *EtcdServer) updateClusterVersion(ver string) {
	if s.cluster.Version() == nil {
		plog.Infof("setting up the initial cluster version to %s", version.Cluster(ver))
	} else {
		plog.Infof("updating the cluster version from %s to %s", version.Cluster(s.cluster.Version().String()), version.Cluster(ver))
	}
	req := pb.Request{
		Method: "PUT",
		Path:   membership.StoreClusterVersionKey(),
		Val:    ver,
	}
	ctx, cancel := context.WithTimeout(context.Background(), s.Cfg.ReqTimeout())
	_, err := s.Do(ctx, req)
	cancel()
	switch err {
	case nil:
		return
	case ErrStopped:
		plog.Infof("aborting update cluster version because server is stopped")
		return
	default:
		plog.Errorf("error updating cluster version (%v)", err)
	}
}

// UpdateCapability updates the enabledMap when the cluster version increases.
func UpdateCapability(v *semver.Version) {
	if v == nil {
		// if recovered but version was never set by cluster
		return
	}
	enableMapMu.Lock()
	if curVersion != nil && !curVersion.LessThan(*v) {
		enableMapMu.Unlock()
		return
	}
	curVersion = v
	enabledMap = capabilityMaps[curVersion.String()]
	enableMapMu.Unlock()
	plog.Infof("enabled capabilities for version %s", version.Cluster(v.String()))
}

func (c *RaftCluster) Recover(onSet func(*semver.Version)) {
	c.Lock()
	defer c.Unlock()

	c.members, c.removed = membersFromStore(c.store)
	c.version = clusterVersionFromStore(c.store)
	mustDetectDowngrade(c.version)
	onSet(c.version)

	for _, m := range c.members {
		plog.Infof("added member %s %v to cluster %s from store", m.ID, m.PeerURLs, c.id)
	}
	if c.version != nil {
		plog.Infof("set the cluster version to %v from store", version.Cluster(c.version.String()))
	}
}

func runCapabilityLoop(s *etcdserver.EtcdServer) {
	stopped := s.StopNotify()

	var pv *semver.Version
	for {
		if v := s.ClusterVersion(); v != pv {
			if pv == nil || (v != nil && pv.LessThan(*v)) {
				pv = v
				enableMapMu.Lock()
				enabledMap = capabilityMaps[pv.String()]
				enableMapMu.Unlock()
				plog.Infof("enabled capabilities for version %s", version.Cluster(pv.String()))
			}
		}

		select {
		case <-stopped:
			return
		case <-time.After(500 * time.Millisecond):
		}
	}
}

// start prepares and starts server in a new goroutine. It is no longer safe to
// modify a server's fields after it has been sent to Start.
// This function is just used for testing.
func (s *EtcdServer) start() {
	if s.snapCount == 0 {
		plog.Infof("set snapshot count to default %d", DefaultSnapCount)
		s.snapCount = DefaultSnapCount
	}
	s.w = wait.New()
	s.done = make(chan struct{})
	s.stop = make(chan struct{})
	if s.ClusterVersion() != nil {
		plog.Infof("starting server... [version: %v, cluster version: %v]", version.Version, version.Cluster(s.ClusterVersion().String()))
	} else {
		plog.Infof("starting server... [version: %v, cluster version: to_be_decided]", version.Version)
	}
	// TODO: if this is an empty log, writes all peer infos
	// into the first entry
	go s.run()
}

func MustDetectDowngrade(cv *semver.Version) {
	lv := semver.Must(semver.NewVersion(version.Version))
	// only keep major.minor version for comparison against cluster version
	lv = &semver.Version{Major: lv.Major, Minor: lv.Minor}
	if cv != nil && lv.LessThan(*cv) {
		plog.Fatalf("cluster cannot be downgraded (current version: %s is lower than determined cluster version: %s).", version.Version, version.Cluster(cv.String()))
	}
}