fix a data race related to query.ackCh (#651)

serf/query.go

@@ -185,6 +185,22 @@ func (r *QueryResponse) sendResponse(nr NodeResponse) error {
 	return nil
 }
 
+// sendResponse sends a response on the response channel ensuring the channel is not closed.
+func (r *QueryResponse) sendAck(nr *messageQueryResponse) error {
+	r.closeLock.Lock()
+	defer r.closeLock.Unlock()
+	if r.closed {
+		return nil
+	}
+	select {
+	case r.ackCh <- nr.From:
+		r.acks[nr.From] = struct{}{}
+	default:
+		return errors.New("serf: Failed to deliver query response, dropping")
+	}
+	return nil
+}
+
 // NodeResponse is used to represent a single response from a node
 type NodeResponse struct {
 	From    string

serf/serf.go

@@ -930,7 +930,7 @@ func (s *Serf) handleNodeJoin(n *memberlist.Node) {
 		member = &memberState{
 			Member: Member{
 				Name:   n.Name,
-				Addr:   net.IP(n.Addr),
+				Addr:   n.Addr,
 				Port:   n.Port,
 				Tags:   s.decodeTags(n.Meta),
 				Status: StatusAlive,
@@ -1056,7 +1056,7 @@ func (s *Serf) handleNodeUpdate(n *memberlist.Node) {
 	}
 
 	// Update the member attributes
-	member.Addr = net.IP(n.Addr)
+	member.Addr = n.Addr
 	member.Port = n.Port
 	member.Tags = s.decodeTags(n.Meta)
 
@@ -1424,11 +1424,9 @@ func (s *Serf) handleQueryResponse(resp *messageQueryResponse) {
 		}
 
 		metrics.IncrCounter([]string{"serf", "query_acks"}, 1)
-		select {
-		case query.ackCh <- resp.From:
-			query.acks[resp.From] = struct{}{}
-		default:
-			s.logger.Printf("[WARN] serf: Failed to deliver query ack, dropping")
+		err := query.sendAck(resp)
+		if err != nil {
+			s.logger.Printf("[WARN] %v", err)
 		}
 	} else {
 		// Exit early if this is a duplicate response