cmd/devp2p/internal/ethtest: add more tx propagation tests (#22630)

This adds a test for large tx announcement messages, as well as a test to
check that announced tx hashes are requested by the node.

Author: renaynay

cmd/devp2p/internal/ethtest/eth66_suite.go

@@ -19,6 +19,7 @@ package ethtest
 import (
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
@@ -125,22 +126,7 @@ func (s *Suite) TestSimultaneousRequests_66(t *utesting.T) {
 // TestBroadcast_66 tests whether a block announcement is correctly
 // propagated to the given node's peer(s) on the eth66 protocol.
 func (s *Suite) TestBroadcast_66(t *utesting.T) {
-	sendConn, receiveConn := s.setupConnection66(t), s.setupConnection66(t)
-	defer sendConn.Close()
-	defer receiveConn.Close()
-
-	nextBlock := len(s.chain.blocks)
-	blockAnnouncement := &NewBlock{
-		Block: s.fullChain.blocks[nextBlock],
-		TD:    s.fullChain.TD(nextBlock + 1),
-	}
-	s.testAnnounce66(t, sendConn, receiveConn, blockAnnouncement)
-	// update test suite chain
-	s.chain.blocks = append(s.chain.blocks, s.fullChain.blocks[nextBlock])
-	// wait for client to update its chain
-	if err := receiveConn.waitForBlock66(s.chain.Head()); err != nil {
-		t.Fatal(err)
-	}
+	s.sendNextBlock66(t)
 }
 
 // TestGetBlockBodies_66 tests whether the given node can respond to
@@ -426,3 +412,82 @@ func (s *Suite) TestSameRequestID_66(t *utesting.T) {
 	// check response from first request
 	headersMatch(t, s.chain, s.getBlockHeaders66(t, conn, req1, reqID))
 }
+
+// TestLargeTxRequest_66 tests whether a node can fulfill a large GetPooledTransactions
+// request.
+func (s *Suite) TestLargeTxRequest_66(t *utesting.T) {
+	// send the next block to ensure the node is no longer syncing and is able to accept
+	// txs
+	s.sendNextBlock66(t)
+	// send 2000 transactions to the node
+	hashMap, txs := generateTxs(t, s, 2000)
+	sendConn := s.setupConnection66(t)
+	defer sendConn.Close()
+
+	sendMultipleSuccessfulTxs(t, s, sendConn, txs)
+	// set up connection to receive to ensure node is peered with the receiving connection
+	// before tx request is sent
+	recvConn := s.setupConnection66(t)
+	defer recvConn.Close()
+	// create and send pooled tx request
+	hashes := make([]common.Hash, 0)
+	for _, hash := range hashMap {
+		hashes = append(hashes, hash)
+	}
+	getTxReq := &eth.GetPooledTransactionsPacket66{
+		RequestId:                   1234,
+		GetPooledTransactionsPacket: hashes,
+	}
+	if err := recvConn.write66(getTxReq, GetPooledTransactions{}.Code()); err != nil {
+		t.Fatalf("could not write to conn: %v", err)
+	}
+	// check that all received transactions match those that were sent to node
+	switch msg := recvConn.waitForResponse(s.chain, timeout, getTxReq.RequestId).(type) {
+	case PooledTransactions:
+		for _, gotTx := range msg {
+			if _, exists := hashMap[gotTx.Hash()]; !exists {
+				t.Fatalf("unexpected tx received: %v", gotTx.Hash())
+			}
+		}
+	default:
+		t.Fatalf("unexpected %s", pretty.Sdump(msg))
+	}
+}
+
+// TestNewPooledTxs_66 tests whether a node will do a GetPooledTransactions
+// request upon receiving a NewPooledTransactionHashes announcement.
+func (s *Suite) TestNewPooledTxs_66(t *utesting.T) {
+	// send the next block to ensure the node is no longer syncing and is able to accept
+	// txs
+	s.sendNextBlock66(t)
+	// generate 50 txs
+	hashMap, _ := generateTxs(t, s, 50)
+	// create new pooled tx hashes announcement
+	hashes := make([]common.Hash, 0)
+	for _, hash := range hashMap {
+		hashes = append(hashes, hash)
+	}
+	announce := NewPooledTransactionHashes(hashes)
+	// send announcement
+	conn := s.setupConnection66(t)
+	defer conn.Close()
+	if err := conn.Write(announce); err != nil {
+		t.Fatalf("could not write to connection: %v", err)
+	}
+	// wait for GetPooledTxs request
+	for {
+		_, msg := conn.readAndServe66(s.chain, timeout)
+		switch msg := msg.(type) {
+		case GetPooledTransactions:
+			if len(msg) != len(hashes) {
+				t.Fatalf("unexpected number of txs requested: wanted %d, got %d", len(hashes), len(msg))
+			}
+			return
+		case *NewPooledTransactionHashes:
+			// ignore propagated txs from old tests
+			continue
+		default:
+			t.Fatalf("unexpected %s", pretty.Sdump(msg))
+		}
+	}
+}

cmd/devp2p/internal/ethtest/eth66_suiteHelpers.go

@@ -111,6 +111,18 @@ func (c *Conn) read66() (uint64, Message) {
 		msg = new(Transactions)
 	case (NewPooledTransactionHashes{}).Code():
 		msg = new(NewPooledTransactionHashes)
+	case (GetPooledTransactions{}.Code()):
+		ethMsg := new(eth.GetPooledTransactionsPacket66)
+		if err := rlp.DecodeBytes(rawData, ethMsg); err != nil {
+			return 0, errorf("could not rlp decode message: %v", err)
+		}
+		return ethMsg.RequestId, GetPooledTransactions(ethMsg.GetPooledTransactionsPacket)
+	case (PooledTransactions{}.Code()):
+		ethMsg := new(eth.PooledTransactionsPacket66)
+		if err := rlp.DecodeBytes(rawData, ethMsg); err != nil {
+			return 0, errorf("could not rlp decode message: %v", err)
+		}
+		return ethMsg.RequestId, PooledTransactions(ethMsg.PooledTransactionsPacket)
 	default:
 		msg = errorf("invalid message code: %d", code)
 	}
@@ -124,6 +136,15 @@ func (c *Conn) read66() (uint64, Message) {
 	return 0, errorf("invalid message: %s", string(rawData))
 }
 
+func (c *Conn) waitForResponse(chain *Chain, timeout time.Duration, requestID uint64) Message {
+	for {
+		id, msg := c.readAndServe66(chain, timeout)
+		if id == requestID {
+			return msg
+		}
+	}
+}
+
 // ReadAndServe serves GetBlockHeaders requests while waiting
 // on another message from the node.
 func (c *Conn) readAndServe66(chain *Chain, timeout time.Duration) (uint64, Message) {
@@ -173,27 +194,33 @@ func (s *Suite) testAnnounce66(t *utesting.T, sendConn, receiveConn *Conn, block
 }
 
 func (s *Suite) waitAnnounce66(t *utesting.T, conn *Conn, blockAnnouncement *NewBlock) {
-	timeout := 20 * time.Second
-	_, msg := conn.readAndServe66(s.chain, timeout)
-	switch msg := msg.(type) {
-	case *NewBlock:
-		t.Logf("received NewBlock message: %s", pretty.Sdump(msg.Block))
-		assert.Equal(t,
-			blockAnnouncement.Block.Header(), msg.Block.Header(),
-			"wrong block header in announcement",
-		)
-		assert.Equal(t,
-			blockAnnouncement.TD, msg.TD,
-			"wrong TD in announcement",
-		)
-	case *NewBlockHashes:
-		blockHashes := *msg
-		t.Logf("received NewBlockHashes message: %s", pretty.Sdump(blockHashes))
-		assert.Equal(t, blockAnnouncement.Block.Hash(), blockHashes[0].Hash,
-			"wrong block hash in announcement",
-		)
-	default:
-		t.Fatalf("unexpected: %s", pretty.Sdump(msg))
+	for {
+		_, msg := conn.readAndServe66(s.chain, timeout)
+		switch msg := msg.(type) {
+		case *NewBlock:
+			t.Logf("received NewBlock message: %s", pretty.Sdump(msg.Block))
+			assert.Equal(t,
+				blockAnnouncement.Block.Header(), msg.Block.Header(),
+				"wrong block header in announcement",
+			)
+			assert.Equal(t,
+				blockAnnouncement.TD, msg.TD,
+				"wrong TD in announcement",
+			)
+			return
+		case *NewBlockHashes:
+			blockHashes := *msg
+			t.Logf("received NewBlockHashes message: %s", pretty.Sdump(blockHashes))
+			assert.Equal(t, blockAnnouncement.Block.Hash(), blockHashes[0].Hash,
+				"wrong block hash in announcement",
+			)
+			return
+		case *NewPooledTransactionHashes:
+			// ignore old txs being propagated
+			continue
+		default:
+			t.Fatalf("unexpected: %s", pretty.Sdump(msg))
+		}
 	}
 }
 
@@ -268,3 +295,24 @@ func headersMatch(t *utesting.T, chain *Chain, headers BlockHeaders) {
 		assert.Equal(t, chain.blocks[int(num)].Header(), header)
 	}
 }
+
+func (s *Suite) sendNextBlock66(t *utesting.T) {
+	sendConn, receiveConn := s.setupConnection66(t), s.setupConnection66(t)
+	defer sendConn.Close()
+	defer receiveConn.Close()
+
+	// create new block announcement
+	nextBlock := len(s.chain.blocks)
+	blockAnnouncement := &NewBlock{
+		Block: s.fullChain.blocks[nextBlock],
+		TD:    s.fullChain.TD(nextBlock + 1),
+	}
+	// send announcement and wait for node to request the header
+	s.testAnnounce66(t, sendConn, receiveConn, blockAnnouncement)
+	// update test suite chain
+	s.chain.blocks = append(s.chain.blocks, s.fullChain.blocks[nextBlock])
+	// wait for client to update its chain
+	if err := receiveConn.waitForBlock66(s.chain.Head()); err != nil {
+		t.Fatal(err)
+	}
+}

cmd/devp2p/internal/ethtest/suite.go

@@ -97,6 +97,8 @@ func (s *Suite) AllEthTests() []utesting.Test {
 		{Name: "TestTransaction_66", Fn: s.TestTransaction_66},
 		{Name: "TestMaliciousTx", Fn: s.TestMaliciousTx},
 		{Name: "TestMaliciousTx_66", Fn: s.TestMaliciousTx_66},
+		{Name: "TestLargeTxRequest_66", Fn: s.TestLargeTxRequest_66},
+		{Name: "TestNewPooledTxs_66", Fn: s.TestNewPooledTxs_66},
 	}
 }
 
@@ -129,6 +131,8 @@ func (s *Suite) Eth66Tests() []utesting.Test {
 		{Name: "TestMaliciousStatus_66", Fn: s.TestMaliciousStatus_66},
 		{Name: "TestTransaction_66", Fn: s.TestTransaction_66},
 		{Name: "TestMaliciousTx_66", Fn: s.TestMaliciousTx_66},
+		{Name: "TestLargeTxRequest_66", Fn: s.TestLargeTxRequest_66},
+		{Name: "TestNewPooledTxs_66", Fn: s.TestNewPooledTxs_66},
 	}
 }
 
@@ -455,7 +459,6 @@ func (s *Suite) testAnnounce(t *utesting.T, sendConn, receiveConn *Conn, blockAn
 }
 
 func (s *Suite) waitAnnounce(t *utesting.T, conn *Conn, blockAnnouncement *NewBlock) {
-	timeout := 20 * time.Second
 	switch msg := conn.ReadAndServe(s.chain, timeout).(type) {
 	case *NewBlock:
 		t.Logf("received NewBlock message: %s", pretty.Sdump(msg.Block))