polcies: dont mark a removed host

Dont mark a host which has been removed from the policy/pools, this
fixes a race where the pool can remove the policies host, then a query
which is using the host will try to mark the host, which does not exist
in the underlying hostpool, causing a log.Fatalf/panic.

Fix this by checking the policy still has the host we want to mark, and
do so whilst holding the mutex read lock so that the hostpool will not
change.

policies.go

@@ -485,15 +485,20 @@ func (r *hostPoolHostPolicy) Pick(qry *Query) NextHost {
 			return nil
 		}
 
-		return selectedHostPoolHost{host, hostR}
+		return selectedHostPoolHost{
+			policy: r,
+			info:   host,
+			hostR:  hostR,
+		}
 	}
 }
 
 // selectedHostPoolHost is a host returned by the hostPoolHostPolicy and
 // implements the SelectedHost interface
 type selectedHostPoolHost struct {
-	info  *HostInfo
-	hostR hostpool.HostPoolResponse
+	policy *hostPoolHostPolicy
+	info   *HostInfo
+	hostR  hostpool.HostPoolResponse
 }
 
 func (host selectedHostPoolHost) Info() *HostInfo {
@@ -501,6 +506,14 @@ func (host selectedHostPoolHost) Info() *HostInfo {
 }
 
 func (host selectedHostPoolHost) Mark(err error) {
+	host.policy.mu.RLock()
+	defer host.policy.mu.RUnlock()
+
+	if _, ok := host.policy.hostMap[host.info.Peer()]; !ok {
+		// host was removed between pick and mark
+		return
+	}
+
 	host.hostR.Mark(err)
 }