[NOD-1444] Implement getHeaders RPC command (#944)

* [NOD-1444] Implement getHeaders RPC command

* [NOD-1444] Fix tests and comments

* [NOD-1444] Fix error message

* [NOD-1444] Make GetHeaders propagate header serialization errors

* [NOD-1444] RLock the dag on GetHeaders

* [NOD-1444] Change the error field number to 1000

app/appmessage/message.go

@@ -91,6 +91,8 @@ const (
 	CmdGetBlockCountResponseMessage
 	CmdGetBlockDAGInfoRequestMessage
 	CmdGetBlockDAGInfoResponseMessage
+	CmdGetHeadersRequestMessage
+	CmdGetHeadersResponseMessage
 )
 
 // ProtocolMessageCommandToString maps all MessageCommands to their string representation
@@ -158,6 +160,8 @@ var RPCMessageCommandToString = map[MessageCommand]string{
 	CmdGetBlockCountResponseMessage:        "GetBlockCountResponse",
 	CmdGetBlockDAGInfoRequestMessage:       "GetBlockDAGInfoRequest",
 	CmdGetBlockDAGInfoResponseMessage:      "GetBlockDAGInfoResponse",
+	CmdGetHeadersRequestMessage:            "GetHeadersRequest",
+	CmdGetHeadersResponseMessage:           "GetHeadersResponse",
 }
 
 // Message is an interface that describes a kaspa message. A type that

app/appmessage/p2p_msgrequestibdblocks_test.go

@@ -18,7 +18,7 @@ func TestRequstIBDBlocks(t *testing.T) {
 		t.Errorf("NewHashFromStr: %v", err)
 	}
 
-	hashStr = "3ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506"
+	hashStr = "000000000003ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506"
 	highHash, err := daghash.NewHashFromStr(hashStr)
 	if err != nil {
 		t.Errorf("NewHashFromStr: %v", err)

app/appmessage/p2p_msgtx_test.go

@@ -23,7 +23,7 @@ import (
 func TestTx(t *testing.T) {
 	pver := ProtocolVersion
 
-	txIDStr := "3ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506"
+	txIDStr := "000000000003ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506"
 	txID, err := daghash.NewTxIDFromStr(txIDStr)
 	if err != nil {
 		t.Errorf("NewTxIDFromStr: %v", err)

app/appmessage/rpc_get_headers.go

@@ -0,0 +1,45 @@
+package appmessage
+
+// GetHeadersRequestMessage is an appmessage corresponding to
+// its respective RPC message
+type GetHeadersRequestMessage struct {
+	baseMessage
+	StartHash   string
+	Limit       uint64
+	IsAscending bool
+}
+
+// Command returns the protocol command string for the message
+func (msg *GetHeadersRequestMessage) Command() MessageCommand {
+	return CmdGetHeadersRequestMessage
+}
+
+// NewGetHeadersRequestMessage returns a instance of the message
+func NewGetHeadersRequestMessage(startHash string, limit uint64, isAscending bool) *GetHeadersRequestMessage {
+	return &GetHeadersRequestMessage{
+		StartHash:   startHash,
+		Limit:       limit,
+		IsAscending: isAscending,
+	}
+}
+
+// GetHeadersResponseMessage is an appmessage corresponding to
+// its respective RPC message
+type GetHeadersResponseMessage struct {
+	baseMessage
+	Headers []string
+
+	Error *RPCError
+}
+
+// Command returns the protocol command string for the message
+func (msg *GetHeadersResponseMessage) Command() MessageCommand {
+	return CmdGetHeadersResponseMessage
+}
+
+// NewGetHeadersResponseMessage returns a instance of the message
+func NewGetHeadersResponseMessage(headers []string) *GetHeadersResponseMessage {
+	return &GetHeadersResponseMessage{
+		Headers: headers,
+	}
+}

app/rpc/rpc.go

@@ -29,6 +29,7 @@ var handlers = map[appmessage.MessageCommand]handler{
 	appmessage.CmdGetBlocksRequestMessage:            rpchandlers.HandleGetBlocks,
 	appmessage.CmdGetBlockCountRequestMessage:        rpchandlers.HandleGetBlockCount,
 	appmessage.CmdGetBlockDAGInfoRequestMessage:      rpchandlers.HandleGetBlockDAGInfo,
+	appmessage.CmdGetHeadersRequestMessage:           rpchandlers.HandleGetHeaders,
 }
 
 func (m *Manager) routerInitializer(router *router.Router, netConnection *netadapter.NetConnection) {

app/rpc/rpchandlers/get_headers.go

@@ -0,0 +1,52 @@
+package rpchandlers
+
+import (
+	"bytes"
+	"encoding/hex"
+	"github.com/kaspanet/kaspad/app/appmessage"
+	"github.com/kaspanet/kaspad/app/rpc/rpccontext"
+	"github.com/kaspanet/kaspad/infrastructure/network/netadapter/router"
+	"github.com/kaspanet/kaspad/util/daghash"
+)
+
+// HandleGetHeaders handles the respectively named RPC command
+func HandleGetHeaders(context *rpccontext.Context, _ *router.Router, request appmessage.Message) (appmessage.Message, error) {
+	getHeadersRequest := request.(*appmessage.GetHeadersRequestMessage)
+	dag := context.DAG
+
+	var startHash *daghash.Hash
+	if getHeadersRequest.StartHash != "" {
+		var err error
+		startHash, err = daghash.NewHashFromStr(getHeadersRequest.StartHash)
+		if err != nil {
+			errorMessage := &appmessage.GetHeadersResponseMessage{}
+			errorMessage.Error = appmessage.RPCErrorf("Start hash could not be parsed: %s", err)
+			return errorMessage, nil
+		}
+	}
+
+	const getHeadersDefaultLimit uint64 = 2000
+	limit := getHeadersDefaultLimit
+	if getHeadersRequest.Limit != 0 {
+		limit = getHeadersRequest.Limit
+	}
+
+	headers, err := dag.GetHeaders(startHash, limit, getHeadersRequest.IsAscending)
+	if err != nil {
+		errorMessage := &appmessage.GetHeadersResponseMessage{}
+		errorMessage.Error = appmessage.RPCErrorf("Error getting the headers: %s", err)
+		return errorMessage, nil
+	}
+
+	headersHex := make([]string, len(headers))
+	var buf bytes.Buffer
+	for i, header := range headers {
+		err := header.Serialize(&buf)
+		if err != nil {
+			return nil, err
+		}
+		headersHex[i] = hex.EncodeToString(buf.Bytes())
+		buf.Reset()
+	}
+	return appmessage.NewGetHeadersResponseMessage(headersHex), nil
+}

domain/blockdag/dag.go

@@ -360,17 +360,30 @@ func (dag *BlockDAG) isInPast(this *blockNode, other *blockNode) (bool, error) {
 	return dag.reachabilityTree.isInPast(this, other)
 }
 
-// GetTopHeaders returns the top appmessage.MaxBlockHeadersPerMsg block headers ordered by blue score.
+// GetHeaders returns DAG headers ordered by blue score, starts from the given hash with the given direction.
-func (dag *BlockDAG) GetTopHeaders(highHash *daghash.Hash, maxHeaders uint64) ([]*appmessage.BlockHeader, error) {
+func (dag *BlockDAG) GetHeaders(startHash *daghash.Hash, maxHeaders uint64,
+	isAscending bool) ([]*appmessage.BlockHeader, error) {
+
+	dag.RLock()
+	defer dag.RUnlock()
+
+	if isAscending {
+		return dag.getHeadersAscending(startHash, maxHeaders)
+	}
+
+	return dag.getHeadersDescending(startHash, maxHeaders)
+}
+
+func (dag *BlockDAG) getHeadersDescending(highHash *daghash.Hash, maxHeaders uint64) ([]*appmessage.BlockHeader, error) {
 	highNode := &dag.virtual.blockNode
 	if highHash != nil {
 		var ok bool
 		highNode, ok = dag.index.LookupNode(highHash)
 		if !ok {
-			return nil, errors.Errorf("Couldn't find the high hash %s in the dag", highHash)
+			return nil, errors.Errorf("Couldn't find the start hash %s in the dag", highHash)
 		}
 	}
-	headers := make([]*appmessage.BlockHeader, 0, highNode.blueScore)
+	headers := make([]*appmessage.BlockHeader, 0, maxHeaders)
 	queue := newDownHeap()
 	queue.pushSet(highNode.parents)
 
@@ -386,6 +399,31 @@ func (dag *BlockDAG) GetTopHeaders(highHash *daghash.Hash, maxHeaders uint64) ([
 	return headers, nil
 }
 
+func (dag *BlockDAG) getHeadersAscending(lowHash *daghash.Hash, maxHeaders uint64) ([]*appmessage.BlockHeader, error) {
+	lowNode := dag.genesis
+	if lowHash != nil {
+		var ok bool
+		lowNode, ok = dag.index.LookupNode(lowHash)
+		if !ok {
+			return nil, errors.Errorf("Couldn't find the start hash %s in the dag", lowHash)
+		}
+	}
+	headers := make([]*appmessage.BlockHeader, 0, maxHeaders)
+	queue := newUpHeap()
+	queue.pushSet(lowNode.children)
+
+	visited := newBlockSet()
+	for i := uint32(0); queue.Len() > 0 && uint64(len(headers)) < maxHeaders; i++ {
+		current := queue.pop()
+		if !visited.contains(current) {
+			visited.add(current)
+			headers = append(headers, current.Header())
+			queue.pushSet(current.children)
+		}
+	}
+	return headers, nil
+}
+
 // ForEachHash runs the given fn on every hash that's currently known to
 // the DAG.
 //

domain/blockdag/dag_test.go

@@ -214,7 +214,7 @@ func TestIsKnownBlock(t *testing.T) {
 		{hash: "732c891529619d43b5aeb3df42ba25dea483a8c0aded1cf585751ebabea28f29", want: true},
 
 		// Random hashes should not be available.
-		{hash: "123", want: false},
+		{hash: "1234567812345678123456781234567812345678123456781234567812345678", want: false},
 	}
 
 	for i, test := range tests {

domain/mempool/mempool_test.go

@@ -627,10 +627,7 @@ func TestProcessTransaction(t *testing.T) {
 		t.Fatalf("Script: error creating wrappedP2shNonSigScript: %v", err)
 	}
 
-	dummyPrevOutTxID, err := daghash.NewTxIDFromStr("01")
+	dummyPrevOutTxID := &daghash.TxID{}
-	if err != nil {
-		t.Fatalf("NewShaHashFromStr: unexpected error: %v", err)
-	}
 	dummyPrevOut := appmessage.Outpoint{TxID: *dummyPrevOutTxID, Index: 1}
 	dummySigScript := bytes.Repeat([]byte{0x00}, 65)
 

domain/mempool/policy_test.go

@@ -168,10 +168,7 @@ func TestDust(t *testing.T) {
 // TestCheckTransactionStandard tests the checkTransactionStandard API.
 func TestCheckTransactionStandard(t *testing.T) {
 	// Create some dummy, but otherwise standard, data for transactions.
-	prevOutTxID, err := daghash.NewTxIDFromStr("01")
+	prevOutTxID := &daghash.TxID{}
-	if err != nil {
-		t.Fatalf("NewShaHashFromStr: unexpected error: %v", err)
-	}
 	dummyPrevOut := appmessage.Outpoint{TxID: *prevOutTxID, Index: 1}
 	dummySigScript := bytes.Repeat([]byte{0x00}, 65)
 	dummyTxIn := appmessage.TxIn{

infrastructure/network/netadapter/server/grpcserver/protowire/messages.proto

@@ -64,6 +64,8 @@ message KaspadMessage {
     GetBlockCountResponseMessage getBlockCountResponse = 1034;
     GetBlockDagInfoRequestMessage getBlockDagInfoRequest = 1035;
     GetBlockDagInfoResponseMessage getBlockDagInfoResponse = 1036;
+    GetHeadersRequestMessage getHeadersRequest = 1037;
+    GetHeadersResponseMessage getHeadersResponse = 1038;
   }
 }
 
@@ -562,6 +564,17 @@ message GetBlockDagInfoResponseMessage{
   RPCError error = 6;
 }
 
+message GetHeadersRequestMessage{
+  string startHash = 1;
+  uint64 limit = 2;
+  bool isAscending = 3;
+}
+
+message GetHeadersResponseMessage{
+  repeated string headers = 1;
+  RPCError error = 1000;
+}
+
 service RPC {
   rpc MessageStream (stream KaspadMessage) returns (stream KaspadMessage) {}
 }

infrastructure/network/netadapter/server/grpcserver/protowire/rpc_get_headers.go

@@ -0,0 +1,45 @@
+package protowire
+
+import (
+	"github.com/kaspanet/kaspad/app/appmessage"
+)
+
+func (x *KaspadMessage_GetHeadersRequest) toAppMessage() (appmessage.Message, error) {
+	return &appmessage.GetHeadersRequestMessage{
+		StartHash:   x.GetHeadersRequest.StartHash,
+		Limit:       x.GetHeadersRequest.Limit,
+		IsAscending: x.GetHeadersRequest.IsAscending,
+	}, nil
+}
+
+func (x *KaspadMessage_GetHeadersRequest) fromAppMessage(message *appmessage.GetHeadersRequestMessage) error {
+	x.GetHeadersRequest = &GetHeadersRequestMessage{
+		StartHash:   message.StartHash,
+		Limit:       message.Limit,
+		IsAscending: message.IsAscending,
+	}
+	return nil
+}
+
+func (x *KaspadMessage_GetHeadersResponse) toAppMessage() (appmessage.Message, error) {
+	var err *appmessage.RPCError
+	if x.GetHeadersResponse.Error != nil {
+		err = &appmessage.RPCError{Message: x.GetHeadersResponse.Error.Message}
+	}
+	return &appmessage.GetHeadersResponseMessage{
+		Headers: x.GetHeadersResponse.Headers,
+		Error:   err,
+	}, nil
+}
+
+func (x *KaspadMessage_GetHeadersResponse) fromAppMessage(message *appmessage.GetHeadersResponseMessage) error {
+	var err *RPCError
+	if message.Error != nil {
+		err = &RPCError{Message: message.Error.Message}
+	}
+	x.GetHeadersResponse = &GetHeadersResponseMessage{
+		Headers: message.Headers,
+		Error:   err,
+	}
+	return nil
+}

infrastructure/network/netadapter/server/grpcserver/protowire/wire.go

@@ -464,6 +464,22 @@ func toRPCPayload(message appmessage.Message) (isKaspadMessage_Payload, error) {
 			return nil, err
 		}
 		return payload, nil
+
+	case *appmessage.GetHeadersRequestMessage:
+		payload := new(KaspadMessage_GetHeadersRequest)
+		err := payload.fromAppMessage(message)
+		if err != nil {
+			return nil, err
+		}
+		return payload, nil
+
+	case *appmessage.GetHeadersResponseMessage:
+		payload := new(KaspadMessage_GetHeadersResponse)
+		err := payload.fromAppMessage(message)
+		if err != nil {
+			return nil, err
+		}
+		return payload, nil
 	default:
 		return nil, nil
 	}

infrastructure/network/rpcclient/rpc_get_headers.go

@@ -0,0 +1,20 @@
+package rpcclient
+
+import "github.com/kaspanet/kaspad/app/appmessage"
+
+// GetHeaders sends an RPC request respective to the function's name and returns the RPC server's response
+func (c *RPCClient) GetHeaders(startHash string, limit uint64, isAscending bool) (*appmessage.GetHeadersResponseMessage, error) {
+	err := c.rpcRouter.outgoingRoute().Enqueue(appmessage.NewGetHeadersRequestMessage(startHash, limit, isAscending))
+	if err != nil {
+		return nil, err
+	}
+	response, err := c.route(appmessage.CmdGetHeadersResponseMessage).DequeueWithTimeout(c.timeout)
+	if err != nil {
+		return nil, err
+	}
+	getHeadersResponse := response.(*appmessage.GetHeadersResponseMessage)
+	if getHeadersResponse.Error != nil {
+		return nil, c.convertRPCError(getHeadersResponse.Error)
+	}
+	return getHeadersResponse, nil
+}

util/daghash/hash.go

@@ -19,12 +19,12 @@ const HashSize = 32
 // TxIDSize of array used to store TxID. See TxID.
 const TxIDSize = HashSize
 
-// MaxHashStringSize is the maximum length of a Hash hash string.
+// HashStringSize is the length of a Hash hash string.
-const MaxHashStringSize = HashSize * 2
+const HashStringSize = HashSize * 2
 
 // ErrHashStrSize describes an error that indicates the caller specified a hash
-// string that has too many characters.
+// string that hasn't the correct number of characters.
-var ErrHashStrSize = errors.Errorf("max hash string length is %d bytes", MaxHashStringSize)
+var ErrHashStrSize = errors.Errorf("hash string length should be %d bytes", HashStringSize)
 
 // Hash is used in several of the kaspa messages and common structures. It
 // typically represents the double sha256 of data.
@@ -172,7 +172,7 @@ func NewTxIDFromStr(idStr string) (*TxID, error) {
 // destination.
 func Decode(dst *Hash, src string) error {
 	// Return error if hash string is too long.
-	if len(src) > MaxHashStringSize {
+	if len(src) != HashStringSize {
 		return ErrHashStrSize
 	}
 

util/daghash/hash_test.go

@@ -26,7 +26,7 @@ var mainnetGenesisHash = Hash([HashSize]byte{
 // TestHash tests the Hash API.
 func TestHash(t *testing.T) {
 	// Hash of block 234439.
-	blockHashStr := "14a0810ac680a3eb3f82edc878cea25ec41d6b790744e5daeef"
+	blockHashStr := "d2f0fb908b59cd20d8687fadca033495d355dccf7718d50e2f9b4826a1f853a8"
 	blockHash, err := NewHashFromStr(blockHashStr)
 	if err != nil {
 		t.Errorf("NewHashFromStr: %v", err)
@@ -158,42 +158,25 @@ func TestNewHashFromStr(t *testing.T) {
 			nil,
 		},
 
-		// Genesis hash with stripped leading zeros.
-		{
-			"63bbcfdd699ffd8cb19878564b14d3af8ba4d7ee4d1dd54925a7c21d5b5b5fdc",
-			mainnetGenesisHash,
-			nil,
-		},
-
 		// Empty string.
 		{
 			"",
 			Hash{},
-			nil,
+			ErrHashStrSize,
 		},
 
 		// Single digit hash.
 		{
 			"1",
-			Hash([HashSize]byte{
+			Hash{},
-				0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			ErrHashStrSize,
-				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-			}),
-			nil,
 		},
 
 		// Block 203707 with stripped leading zeros.
 		{
 			"3264bc2ac36a60840790ba1d475d01367e7c723da941069e9dc",
-			Hash([HashSize]byte{
+			Hash{},
-				0xdc, 0xe9, 0x69, 0x10, 0x94, 0xda, 0x23, 0xc7,
+			ErrHashStrSize,
-				0xe7, 0x67, 0x13, 0xd0, 0x75, 0xd4, 0xa1, 0x0b,
-				0x79, 0x40, 0x08, 0xa6, 0x36, 0xac, 0xc2, 0x4b,
-				0x26, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-			}),
-			nil,
 		},
 
 		// Hash string that is too long.
@@ -205,7 +188,7 @@ func TestNewHashFromStr(t *testing.T) {
 
 		// Hash string that is contains non-hex chars.
 		{
-			"abcdefg",
+			"abcdefgggggggggggggggggggggggggggggggggggggggggggggggggggggggggg",
 			Hash{},
 			hex.InvalidByteError('g'),
 		},