fix(registry): fetch peers from store instead of cache

The current cache implmentation may contain removed machines, so we
fetch peers from store for correctness.

server/peer_server.go

@@ -737,8 +737,8 @@ func (s *PeerServer) monitorActiveSize() {
 
 		// Retrieve target active size and actual active size.
 		activeSize := s.ClusterConfig().ActiveSize
-		peerCount := s.registry.Count()
 		peers := s.registry.Names()
+		peerCount := s.registry.Count()
 		if index := sort.SearchStrings(peers, s.Config.Name); index < len(peers) && peers[index] == s.Config.Name {
 			peers = append(peers[:index], peers[index+1:]...)
 		}

server/registry.go

@@ -38,19 +38,6 @@ func NewRegistry(s store.Store) *Registry {
 	}
 }
 
-// Names returns a list of cached peer names.
-func (r *Registry) Names() []string {
-	r.Lock()
-	defer r.Unlock()
-
-	names := make([]string, 0, len(r.peers))
-	for name := range r.peers {
-		names = append(names, name)
-	}
-	sort.Sort(sort.StringSlice(names))
-	return names
-}
-
 // Register adds a peer to the registry.
 func (r *Registry) Register(name string, peerURL string, machURL string) error {
 	// Write data to store.
@@ -167,6 +154,17 @@ func (r *Registry) UpdatePeerURL(name string, peerURL string) error {
 	return nil
 }
 
+func (r *Registry) name(key, name string) (string, bool) {
+	return name, true
+}
+
+// Names returns a list of cached peer names.
+func (r *Registry) Names() []string {
+	names := r.urls(RegistryKey, "", "", r.name)
+	sort.Sort(sort.StringSlice(names))
+	return names
+}
+
 // Retrieves the Client URLs for all nodes.
 func (r *Registry) ClientURLs(leaderName, selfName string) []string {
 	return r.urls(RegistryKey, leaderName, selfName, r.clientURL)