hostsource: remove hosts no longer in peers

When refreshing the ring remove hosts from the drivers ring cache to
prevent it from trying to connect to removed hosts if it does not
receive a down host event.

host_source.go

@@ -659,15 +659,24 @@ func (r *ringDescriber) refreshRing() error {
 		return err
 	}
 
+	prevHosts := r.session.ring.currentHosts()
+
 	// TODO: move this to session
-	// TODO: handle removing hosts here
 	for _, h := range hosts {
 		if host, ok := r.session.ring.addHostIfMissing(h); !ok {
 			r.session.pool.addHost(h)
 			r.session.policy.AddHost(h)
 		} else {
 			host.update(h)
 		}
+		delete(prevHosts, h.ConnectAddress().String())
+	}
+
+	// TODO(zariel): it may be worth having a mutex covering the overall ring state
+	// in a session so that everything sees a consistent state. Becuase as is today
+	// events can come in and due to ordering an UP host could be removed from the cluster
+	for _, host := range prevHosts {
+		r.session.removeHost(host)
 	}
 
 	r.session.metadata.setPartitioner(partitioner)

session.go

@@ -384,6 +384,12 @@ func (s *Session) executeQuery(qry *Query) *Iter {
 	return iter
 }
 
+func (s *Session) removeHost(h *HostInfo) {
+	s.policy.RemoveHost(h)
+	s.pool.removeHost(h.ConnectAddress())
+	s.ring.removeHost(h.ConnectAddress())
+}
+
 // KeyspaceMetadata returns the schema metadata for the keyspace specified. Returns an error if the keyspace does not exist.
 func (s *Session) KeyspaceMetadata(keyspace string) (*KeyspaceMetadata, error) {
 	// fail fast