[DVT-1060] Add eth/67 and eth/68 protocols (#143)

cmd/p2p/sensor/sensor.go

@@ -169,7 +169,7 @@ var SensorCmd = &cobra.Command{
 			Number:          block.Number.ToUint64(),
 		}
 
-		opts := p2p.Eth66ProtocolOptions{
+		opts := p2p.EthProtocolOptions{
 			Context:     cmd.Context(),
 			Database:    db,
 			Genesis:     &inputSensorParams.genesis,
@@ -190,11 +190,15 @@ var SensorCmd = &cobra.Command{
 			MaxPeers:       inputSensorParams.MaxPeers,
 			ListenAddr:     fmt.Sprintf(":%d", inputSensorParams.Port),
 			DiscAddr:       fmt.Sprintf(":%d", inputSensorParams.DiscoveryPort),
-			Protocols:      []ethp2p.Protocol{p2p.NewEth66Protocol(opts)},
 			DialRatio:      inputSensorParams.DialRatio,
 			NAT:            inputSensorParams.nat,
 			DiscoveryV4:    true,
 			DiscoveryV5:    true,
+			Protocols: []ethp2p.Protocol{
+				p2p.NewEthProtocol(66, opts),
+				p2p.NewEthProtocol(67, opts),
+				p2p.NewEthProtocol(68, opts),
+			},
 		}
 
 		if inputSensorParams.QuickStart {

p2p/protocol.go

@@ -45,8 +45,8 @@ type conn struct {
 	oldestBlock *types.Header
 }
 
-// Eth66ProtocolOptions is the options used when creating a new eth66 protocol.
-type Eth66ProtocolOptions struct {
+// EthProtocolOptions is the options used when creating a new eth protocol.
+type EthProtocolOptions struct {
 	Context     context.Context
 	Database    database.Database
 	Genesis     *core.Genesis
@@ -70,12 +70,12 @@ type HeadBlock struct {
 	Number          uint64
 }
 
-// NewEth66Proctocol creates the new eth66 protocol. This will handle writing the
+// NewEthProctocol creates the new eth protocol. This will handle writing the
 // status exchange, message handling, and writing blocks/txs to the database.
-func NewEth66Protocol(opts Eth66ProtocolOptions) ethp2p.Protocol {
+func NewEthProtocol(version uint, opts EthProtocolOptions) ethp2p.Protocol {
 	return ethp2p.Protocol{
 		Name:    "eth",
-		Version: 66,
+		Version: version,
 		Length:  17,
 		Run: func(p *ethp2p.Peer, rw ethp2p.MsgReadWriter) error {
 			c := conn{
@@ -134,7 +134,7 @@ func NewEth66Protocol(opts Eth66ProtocolOptions) ethp2p.Protocol {
 				case eth.NewBlockMsg:
 					err = c.handleNewBlock(ctx, msg)
 				case eth.NewPooledTransactionHashesMsg:
-					err = c.handleNewPooledTransactionHashes(ctx, msg)
+					err = c.handleNewPooledTransactionHashes(ctx, version, msg)
 				case eth.GetPooledTransactionsMsg:
 					err = c.handleGetPooledTransactions(msg)
 				case eth.PooledTransactionsMsg:
@@ -412,20 +412,32 @@ func (c *conn) handleGetPooledTransactions(msg ethp2p.Msg) error {
 		&eth.PooledTransactionsPacket66{RequestId: request.RequestId})
 }
 
-func (c *conn) handleNewPooledTransactionHashes(ctx context.Context, msg ethp2p.Msg) error {
-	var txs eth.NewPooledTransactionHashesPacket66
-	if err := msg.Decode(&txs); err != nil {
-		return err
+func (c *conn) handleNewPooledTransactionHashes(ctx context.Context, version uint, msg ethp2p.Msg) error {
+	var hashes []common.Hash
+
+	switch version {
+	case 66, 67:
+		var txs eth.NewPooledTransactionHashesPacket66
+		if err := msg.Decode(&txs); err != nil {
+			return err
+		}
+		hashes = txs
+	case 68:
+		var txs eth.NewPooledTransactionHashesPacket68
+		if err := msg.Decode(&txs); err != nil {
+			return err
+		}
+		hashes = txs.Hashes
+	default:
+		return errors.New("protocol version not found")
 	}
 
-	atomic.AddInt32(&c.count.TransactionHashes, int32(len(txs)))
+	atomic.AddInt32(&c.count.TransactionHashes, int32(len(hashes)))
 
 	if !c.db.ShouldWriteTransactions() || !c.db.ShouldWriteTransactionEvents() {
 		return nil
 	}
 
-	var hashes []common.Hash = txs
-
 	return ethp2p.Send(
 		c.rw,
 		eth.GetPooledTransactionsMsg,

p2p/types.go

@@ -156,7 +156,7 @@ type rlpxConn struct {
 	logger zerolog.Logger
 }
 
-// Read reads an eth66 packet from the connection.
+// Read reads an eth protocol packet from the connection.
 func (c *rlpxConn) Read() Message {
 	code, rawData, _, err := c.Conn.Read()
 	if err != nil {