neutrino + query: Request made more flexible to querying.

This commit makes the query package more flexible
in  querying peers. This is done by adding sendQuery function field to
query.Request struct. Instead of using only QueueMessageWithEncoding
for all requests. This would be useful in coming commits where
we would be using pushGetHeadersMsg to fetch block headers from peers.

Consequent changes:
encoding was removed in the queryJob field as it would not be useful
for all requests. Requests that need encoding can define it as one of
the fields in its own implementation of the new interface created as
a type for the Req field in Request.

PriorityIndex function signature would be used in the coming commits
to indicate the the priority a request should preferably have in a
query batch.

An implementaion of the interface was created for GetCfheaders,
GetCfilter and getData requests. Tests were updated and added to
reflect these changes.

Signed-off-by: Maureen Ononiwu <[email protected]>

Author: Chinwendu20

blockmanager.go

@@ -5,6 +5,7 @@ package neutrino
 import (
 	"bytes"
 	"container/list"
+	"errors"
 	"fmt"
 	"math"
 	"math/big"
@@ -808,19 +809,39 @@ func (b *blockManager) getUncheckpointedCFHeaders(
 // handle a query for checkpointed filter headers.
 type checkpointedCFHeadersQuery struct {
 	blockMgr    *blockManager
-	msgs        []wire.Message
+	msgs        []*encodedQuery
 	checkpoints []*chainhash.Hash
 	stopHashes  map[chainhash.Hash]uint32
 	headerChan  chan *wire.MsgCFHeaders
 }
 
+// encodedQuery holds all the information needed to query a message that pushes requests
+// using the QueryMessagingWithEncoding method.
+type encodedQuery struct {
+	message       wire.Message
+	encoding      wire.MessageEncoding
+	priorityIndex uint64
+}
+
+// Message returns the wire.Message of encodedQuery's struct.
+func (e *encodedQuery) Message() wire.Message {
+	return e.message
+}
+
+// PriorityIndex returns the specified priority the caller wants
+// the request to take.
+func (e *encodedQuery) PriorityIndex() uint64 {
+	return e.priorityIndex
+}
+
 // requests creates the query.Requests for this CF headers query.
 func (c *checkpointedCFHeadersQuery) requests() []*query.Request {
 	reqs := make([]*query.Request, len(c.msgs))
 	for idx, m := range c.msgs {
 		reqs[idx] = &query.Request{
 			Req:        m,
 			HandleResp: c.handleResponse,
+			SendQuery:  sendQueryMessageWithEncoding,
 		}
 	}
 	return reqs
@@ -924,6 +945,24 @@ func (c *checkpointedCFHeadersQuery) handleResponse(req, resp wire.Message,
 	}
 }
 
+// sendQueryMessageWithEncoding sends a message to the peer with encoding.
+func sendQueryMessageWithEncoding(peer query.Peer, req query.ReqMessage) error {
+	sp, ok := peer.(*ServerPeer)
+	if !ok {
+		err := "peer is not of type ServerPeer"
+		log.Errorf(err)
+		return errors.New(err)
+	}
+	request, ok := req.(*encodedQuery)
+	if !ok {
+		return errors.New("invalid request type")
+	}
+
+	sp.QueueMessageWithEncoding(request.message, nil, request.encoding)
+
+	return nil
+}
+
 // getCheckpointedCFHeaders catches a filter header store up with the
 // checkpoints we got from the network. It assumes that the filter header store
 // matches the checkpoints up to the tip of the store.
@@ -959,7 +998,7 @@ func (b *blockManager) getCheckpointedCFHeaders(checkpoints []*chainhash.Hash,
 	// the remaining checkpoint intervals.
 	numCheckpts := uint32(len(checkpoints)) - startingInterval
 	numQueries := (numCheckpts + maxCFCheckptsPerQuery - 1) / maxCFCheckptsPerQuery
-	queryMsgs := make([]wire.Message, 0, numQueries)
+	queryMsgs := make([]*encodedQuery, 0, numQueries)
 
 	// We'll also create an additional set of maps that we'll use to
 	// re-order the responses as we get them in.
@@ -1004,9 +1043,12 @@ func (b *blockManager) getCheckpointedCFHeaders(checkpoints []*chainhash.Hash,
 
 		// Once we have the stop hash, we can construct the query
 		// message itself.
-		queryMsg := wire.NewMsgGetCFHeaders(
-			fType, startHeightRange, &stopHash,
-		)
+		queryMsg := &encodedQuery{
+			message: wire.NewMsgGetCFHeaders(
+				fType, startHeightRange, &stopHash,
+			),
+			encoding: wire.WitnessEncoding,
+		}
 
 		// We'll mark that the ith interval is queried by this message,
 		// and also map the stop hash back to the index of this message.

blockmanager_test.go

@@ -352,7 +352,7 @@ func TestBlockManagerInitialInterval(t *testing.T) {
 
 			var msgs []wire.Message
 			for _, q := range requests {
-				msgs = append(msgs, q.Req)
+				msgs = append(msgs, q.Req.Message())
 			}
 
 			responses, err := generateResponses(msgs, headers)
@@ -582,7 +582,7 @@ func TestBlockManagerInvalidInterval(t *testing.T) {
 
 			var msgs []wire.Message
 			for _, q := range requests {
-				msgs = append(msgs, q.Req)
+				msgs = append(msgs, q.Req.Message())
 			}
 			responses, err := generateResponses(msgs, headers)
 			require.NoError(t, err)

query.go

@@ -435,13 +435,17 @@ type cfiltersQuery struct {
 // request couples a query message with the handler to be used for the response
 // in a query.Request struct.
 func (q *cfiltersQuery) request() *query.Request {
-	msg := wire.NewMsgGetCFilters(
-		q.filterType, uint32(q.startHeight), q.stopHash,
-	)
+	msg := &encodedQuery{
+		message: wire.NewMsgGetCFilters(
+			q.filterType, uint32(q.startHeight), q.stopHash,
+		),
+		encoding: wire.WitnessEncoding,
+	}
 
 	return &query.Request{
 		Req:        msg,
 		HandleResp: q.handleResponse,
+		SendQuery:  sendQueryMessageWithEncoding,
 	}
 }
 
@@ -833,6 +837,10 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 	// Construct the appropriate getdata message to fetch the target block.
 	getData := wire.NewMsgGetData()
 	_ = getData.AddInvVect(inv)
+	msg := &encodedQuery{
+		message:  getData,
+		encoding: wire.WitnessEncoding,
+	}
 
 	var foundBlock *btcutil.Block
 
@@ -912,8 +920,9 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 
 	// Prepare the query request.
 	request := &query.Request{
-		Req:        getData,
+		Req:        msg,
 		HandleResp: handleResp,
+		SendQuery:  sendQueryMessageWithEncoding,
 	}
 
 	// Prepare the query options.

query/interface.go

@@ -125,7 +125,7 @@ type Progress struct {
 // connected peers.
 type Request struct {
 	// Req is the message request to send.
-	Req wire.Message
+	Req ReqMessage
 
 	// HandleResp is a response handler that will be called for every
 	// message received from the peer that the request was made to. It
@@ -139,6 +139,15 @@ type Request struct {
 	// The response should be handed off to another goroutine for
 	// processing.
 	HandleResp func(req, resp wire.Message, peer string) Progress
+
+	// SendQuery handles sending request to the worker's peer. It returns an error,
+	// if one is encountered while sending the request.
+	SendQuery func(peer Peer, request ReqMessage) error
+}
+
+type ReqMessage interface {
+	Message() wire.Message
+	PriorityIndex() uint64
 }
 
 // WorkManager defines an API for a manager that dispatches queries to bitcoin

query/worker.go

@@ -3,8 +3,6 @@ package query
 import (
 	"errors"
 	"time"
-
-	"github.com/btcsuite/btcd/wire"
 )
 
 var (
@@ -27,7 +25,6 @@ type queryJob struct {
 	tries      uint8
 	index      uint64
 	timeout    time.Duration
-	encoding   wire.MessageEncoding
 	cancelChan <-chan struct{}
 	*Request
 }
@@ -89,6 +86,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 	msgChan, cancel := peer.SubscribeRecvMsg()
 	defer cancel()
 
+nexJobLoop:
 	for {
 		log.Tracef("Worker %v waiting for more work", peer.Addr())
 
@@ -133,7 +131,22 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 			log.Tracef("Worker %v queuing job %T with index %v",
 				peer.Addr(), job.Req, job.Index())
 
-			peer.QueueMessageWithEncoding(job.Req, nil, job.encoding)
+			err := job.SendQuery(peer, job.Req)
+
+			// If any error occurs while sending query send a message to the worker's feedback Chan
+			// which would be handled by the "LOOP" below.
+			if err != nil {
+				select {
+				case results <- &jobResult{
+					job:  job,
+					peer: peer,
+					err:  err,
+				}:
+				case <-quit:
+					return
+				}
+				goto nexJobLoop
+			}
 		}
 
 		// Wait for the correct response to be received from the peer,
@@ -143,15 +156,15 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 			timeout = time.NewTimer(job.timeout)
 		)
 
-	Loop:
+	feedbackLoop:
 		for {
 			select {
 			// A message was received from the peer, use the
 			// response handler to check whether it was answering
 			// our request.
 			case resp := <-msgChan:
 				progress := job.HandleResp(
-					job.Req, resp, peer.Addr(),
+					job.Req.Message(), resp, peer.Addr(),
 				)
 
 				log.Tracef("Worker %v handled msg %T while "+
@@ -176,12 +189,12 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 							job.timeout,
 						)
 					}
-					continue Loop
+					continue feedbackLoop
 				}
 
 				// We did get a valid response, and can break
 				// the loop.
-				break Loop
+				break feedbackLoop
 
 			// If the timeout is reached before a valid response
 			// has been received, we exit with an error.
@@ -193,7 +206,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 					"with job index %v", peer.Addr(),
 					job.Req, job.Index())
 
-				break Loop
+				break feedbackLoop
 
 			// If the peer disconnects before giving us a valid
 			// answer, we'll also exit with an error.
@@ -203,7 +216,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 					job.Index())
 
 				jobErr = ErrPeerDisconnected
-				break Loop
+				break feedbackLoop
 
 			// If the job was canceled, we report this back to the
 			// work manager.
@@ -212,7 +225,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 					peer.Addr(), job.Index())
 
 				jobErr = ErrJobCanceled
-				break Loop
+				break feedbackLoop
 
 			case <-quit:
 				return