feat: query witnesses

Author: minhd-vu

p2p/log.go

@@ -21,6 +21,10 @@ type MessageCount struct {
 	Pings               int64 `json:"pings,omitempty"`
 	Errors              int64 `json:"errors,omitempty"`
 	Disconnects         int64 `json:"disconnects,omitempty"`
+	NewWitness          int64 `json:"new_witness,omitempty"`
+	NewWitnessHashes    int64 `json:"new_witness_hashes,omitempty"`
+	GetWitnessRequest   int64 `json:"get_witness_request,omitempty"`
+	Witness             int64 `json:"witness,omitempty"`
 }
 
 // Load takes a snapshot of all the counts in a thread-safe manner. Make sure

p2p/rlpx.go

@@ -37,6 +37,7 @@ func Dial(n *enode.Node) (*rlpxConn, error) {
 			{Name: "eth", Version: 66},
 			{Name: "eth", Version: 67},
 			{Name: "eth", Version: 68},
+			{Name: "wit", Version: 1},
 		},
 	}
 
@@ -194,6 +195,7 @@ func (c *rlpxConn) ReadAndServe(count *MessageCount) error {
 				atomic.AddInt64(&count.BlockHashes, int64(len(*msg)))
 				c.logger.Trace().Msgf("Received %v NewBlockHashes", len(*msg))
 
+				var hashes []common.Hash
 				for _, hash := range *msg {
 					headersRequest := &GetBlockHeaders{
 						GetBlockHeadersRequest: &eth.GetBlockHeadersRequest{
@@ -215,11 +217,30 @@ func (c *rlpxConn) ReadAndServe(count *MessageCount) error {
 					if err := c.Write(bodiesRequest); err != nil {
 						c.logger.Error().Err(err).Msg("Failed to write GetBlockBodies request")
 					}
+
+					hashes = append(hashes, hash.Hash)
 				}
 
+				req := GetWitnessPacket{
+					GetWitnessRequest: &GetWitnessRequest{
+						Hashes: hashes,
+					},
+				}
+				if err := c.Write(req); err != nil {
+					log.Error().Err(err).Msg("Failed to write GetWitnessPacket request")
+				}
 			case *NewBlock:
 				atomic.AddInt64(&count.Blocks, 1)
 				c.logger.Trace().Str("hash", msg.Block.Hash().Hex()).Msg("Received NewBlock")
+
+				req := GetWitnessPacket{
+					GetWitnessRequest: &GetWitnessRequest{
+						Hashes: []common.Hash{msg.Block.Hash()},
+					},
+				}
+				if err := c.Write(req); err != nil {
+					log.Error().Err(err).Msg("Failed to write GetWitnessPacket request")
+				}
 			case *Transactions:
 				atomic.AddInt64(&count.Transactions, int64(len(*msg)))
 				c.logger.Trace().Msgf("Received %v Transactions", len(*msg))
@@ -248,15 +269,27 @@ func (c *rlpxConn) ReadAndServe(count *MessageCount) error {
 				atomic.AddInt64(&count.Errors, 1)
 				c.logger.Trace().Err(msg.Unwrap()).Msg("Received Error")
 
-				if !strings.Contains(msg.Error(), "timeout") {
-					return msg.Unwrap()
-				}
+				// if !strings.Contains(msg.Error(), "timeout") {
+				// 	return msg.Unwrap()
+				// }
 			case *Disconnect:
 				atomic.AddInt64(&count.Disconnects, 1)
 				c.logger.Debug().Msgf("Disconnect received: %v", msg)
 			case *Disconnects:
 				atomic.AddInt64(&count.Disconnects, 1)
 				c.logger.Debug().Msgf("Disconnect received: %v", msg)
+			case *NewWitnessPacket:
+				atomic.AddInt64(&count.NewWitness, 1)
+				c.logger.Debug().Any("msg", msg).Msg("Received NewWitness")
+			case *NewWitnessHashesPacket:
+				atomic.AddInt64(&count.NewWitnessHashes, 1)
+				c.logger.Debug().Any("msg", msg).Msg("Received NewWitnessHashes")
+			case *GetWitnessPacket:
+				atomic.AddInt64(&count.GetWitnessRequest, 1)
+				c.logger.Debug().Any("msg", msg).Msg("Received GetWitnessRequest")
+			case *WitnessPacketRLPPacket:
+				atomic.AddInt64(&count.Witness, 1)
+				c.logger.Debug().Any("msg", msg).Msg("Received Witness")
 			default:
 				c.logger.Info().Interface("msg", msg).Int("code", msg.Code()).Msg("Received message")
 			}

p2p/types.go

@@ -6,13 +6,15 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/stateless"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/rlpx"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 )
 
 type Message interface {
@@ -229,7 +231,16 @@ func (c *rlpxConn) Read() Message {
 			return errorf("could not rlp decode message: %v", err)
 		}
 		return (*PooledTransactions)(ethMsg)
+	case (NewWitnessPacket{}.Code()):
+		msg = new(NewWitnessPacket)
+	case (NewWitnessHashesPacket{}.Code()):
+		msg = new(NewWitnessHashesPacket)
+	case (GetWitnessPacket{}.Code()):
+		msg = new(GetWitnessPacket)
+	case (WitnessPacketRLPPacket{}.Code()):
+		msg = new(WitnessPacketRLPPacket)
 	default:
+		log.Info().Any("code", code).Send()
 		msg = errorf("invalid message code: %d", code)
 	}
 
@@ -331,3 +342,45 @@ type TrieNodes snap.TrieNodesPacket
 
 func (msg TrieNodes) Code() int     { return 40 }
 func (msg TrieNodes) ReqID() uint64 { return msg.ID }
+
+type NewWitnessPacket struct {
+	Witness *stateless.Witness
+}
+
+func (msg NewWitnessPacket) Code() int     { return 33 }
+func (msg NewWitnessPacket) ReqID() uint64 { return 0 }
+
+type NewWitnessHashesPacket struct {
+	Hashes  []common.Hash
+	Numbers []uint64
+}
+
+func (msg NewWitnessHashesPacket) Code() int     { return 34 }
+func (msg NewWitnessHashesPacket) ReqID() uint64 { return 0 }
+
+// GetWitnessRequest represents a list of witnesses query by block hashes.
+type GetWitnessRequest struct {
+	Hashes []common.Hash // Request by list of block hashes
+}
+
+// GetWitnessPacket represents a witness query with request ID wrapping.
+type GetWitnessPacket struct {
+	RequestId uint64
+	*GetWitnessRequest
+}
+
+func (msg GetWitnessPacket) Code() int     { return 35 }
+func (msg GetWitnessPacket) ReqID() uint64 { return msg.RequestId }
+
+// WitnessPacketRLPPacket represents a witness response with request ID wrapping.
+type WitnessPacketRLPPacket struct {
+	RequestId uint64
+	WitnessPacketResponse
+}
+
+// WitnessPacketResponse represents a witness response, to use when we already
+// have the witness rlp encoded.
+type WitnessPacketResponse []rlp.RawValue
+
+func (msg WitnessPacketRLPPacket) Code() int     { return 36 }
+func (msg WitnessPacketRLPPacket) ReqID() uint64 { return msg.RequestId }