eth, eth/downloader: support receipt retrieval (eth/63)

eth/downloader/downloader.go

@@ -46,7 +46,8 @@ var (
 	MaxHashFetch     = 512 // Amount of hashes to be fetched per retrieval request
 	MaxBlockFetch    = 128 // Amount of blocks to be fetched per retrieval request
 	MaxHeaderFetch   = 256 // Amount of block headers to be fetched per retrieval request
-	MaxStateFetch  = 384 // Amount of node
+	MaxStateFetch    = 384 // Amount of node state values to allow fetching per request
+	MaxReceiptsFetch = 384 // Amount of transaction receipts to allow fetching per request
 
 	hashTTL         = 5 * time.Second  // Time it takes for a hash request to time out
 	blockSoftTTL    = 3 * time.Second  // Request completion threshold for increasing or decreasing a peer's bandwidth

eth/handler.go

@@ -59,7 +59,8 @@ func (ep extProt) GetBlock(hashes []common.Hash) error { return ep.getBlocks(has
 type ProtocolManager struct {
 	txpool   txPool
 	chainman *core.ChainManager
-	statedb    common.Database
+	chaindb  common.Database
+
 	downloader *downloader.Downloader
 	fetcher    *fetcher.Fetcher
 	peers      *peerSet
@@ -83,13 +84,13 @@ type ProtocolManager struct {
 
 // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
 // with the ethereum network.
-func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, statedb common.Database) *ProtocolManager {
+func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, chaindb common.Database) *ProtocolManager {
 	// Create the protocol manager with the base fields
 	manager := &ProtocolManager{
 		eventMux:  mux,
 		txpool:    txpool,
 		chainman:  chainman,
-		statedb:   statedb,
+		chaindb:   chaindb,
 		peers:     newPeerSet(),
 		newPeerCh: make(chan *peer, 1),
 		txsyncCh:  make(chan *txsync),
@@ -382,13 +383,40 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				return errResp(ErrDecode, "msg %v: %v", msg, err)
 			}
 			// Retrieve the requested state entry, stopping if enough was found
-			if entry, err := pm.statedb.Get(hash.Bytes()); err == nil {
+			if entry, err := pm.chaindb.Get(hash.Bytes()); err == nil {
 				data = append(data, entry)
 				bytes += len(entry)
 			}
 		}
 		return p.SendNodeData(data)
 
+	case p.version == eth63 && msg.Code == GetReceiptsMsg:
+		// Decode the retrieval message
+		msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
+		if _, err := msgStream.List(); err != nil {
+			return err
+		}
+		// Gather state data until the fetch or network limits is reached
+		var (
+			hash     common.Hash
+			bytes    int
+			receipts []*types.Receipt
+		)
+		for bytes < softResponseLimit && len(receipts) < downloader.MaxReceiptsFetch {
+			// Retrieve the hash of the next transaction receipt
+			if err := msgStream.Decode(&hash); err == rlp.EOL {
+				break
+			} else if err != nil {
+				return errResp(ErrDecode, "msg %v: %v", msg, err)
+			}
+			// Retrieve the requested receipt, stopping if enough was found
+			if receipt := core.GetReceipt(pm.chaindb, hash); receipt != nil {
+				receipts = append(receipts, receipt)
+				bytes += len(receipt.RlpEncode())
+			}
+		}
+		return p.SendReceipts(receipts)
+
 	case msg.Code == NewBlockHashesMsg:
 		// Retrieve and deseralize the remote new block hashes notification
 		var hashes []common.Hash

eth/handler_test.go

@@ -283,9 +283,9 @@ func TestGetNodeData63(t *testing.T) {
 	peer, _ := newTestPeer("peer", 63, pm, true)
 	defer peer.close()
 
-	// Fetch for now the entire state db
+	// Fetch for now the entire chain db
 	hashes := []common.Hash{}
-	for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() {
+	for _, key := range pm.chaindb.(*ethdb.MemDatabase).Keys() {
 		hashes = append(hashes, common.BytesToHash(key))
 	}
 	p2p.Send(peer.app, 0x0d, hashes)
@@ -327,3 +327,62 @@ func TestGetNodeData63(t *testing.T) {
 		}
 	}
 }
+
+// Tests that the transaction receipts can be retrieved based on hashes.
+func TestGetReceipts63(t *testing.T) {
+	// Define three accounts to simulate transactions with
+	acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a")
+	acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee")
+	acc1Addr := crypto.PubkeyToAddress(acc1Key.PublicKey)
+	acc2Addr := crypto.PubkeyToAddress(acc2Key.PublicKey)
+
+	// Create a chain generator with some simple transactions (blatantly stolen from @fjl/chain_makerts_test)
+	generator := func(i int, block *core.BlockGen) {
+		switch i {
+		case 0:
+			// In block 1, the test bank sends account #1 some ether.
+			tx, _ := types.NewTransaction(block.TxNonce(testBankAddress), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil).SignECDSA(testBankKey)
+			block.AddTx(tx)
+		case 1:
+			// In block 2, the test bank sends some more ether to account #1.
+			// acc1Addr passes it on to account #2.
+			tx1, _ := types.NewTransaction(block.TxNonce(testBankAddress), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(testBankKey)
+			tx2, _ := types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(acc1Key)
+			block.AddTx(tx1)
+			block.AddTx(tx2)
+		case 2:
+			// Block 3 is empty but was mined by account #2.
+			block.SetCoinbase(acc2Addr)
+			block.SetExtra([]byte("yeehaw"))
+		case 3:
+			// Block 4 includes blocks 2 and 3 as uncle headers (with modified extra data).
+			b2 := block.PrevBlock(1).Header()
+			b2.Extra = []byte("foo")
+			block.AddUncle(b2)
+			b3 := block.PrevBlock(2).Header()
+			b3.Extra = []byte("foo")
+			block.AddUncle(b3)
+		}
+	}
+	// Assemble the test environment
+	pm := newTestProtocolManager(4, generator, nil)
+	peer, _ := newTestPeer("peer", 63, pm, true)
+	defer peer.close()
+
+	// Collect the hashes to request, and the response to expect
+	hashes := []common.Hash{}
+	for i := uint64(0); i <= pm.chainman.CurrentBlock().NumberU64(); i++ {
+		for _, tx := range pm.chainman.GetBlockByNumber(i).Transactions() {
+			hashes = append(hashes, tx.Hash())
+		}
+	}
+	receipts := make([]*types.Receipt, len(hashes))
+	for i, hash := range hashes {
+		receipts[i] = core.GetReceipt(pm.chaindb, hash)
+	}
+	// Send the hash request and verify the response
+	p2p.Send(peer.app, 0x0f, hashes)
+	if err := p2p.ExpectMsg(peer.app, 0x10, receipts); err != nil {
+		t.Errorf("receipts mismatch: %v", err)
+	}
+}

eth/peer.go

@@ -171,6 +171,12 @@ func (p *peer) SendNodeData(data [][]byte) error {
 	return p2p.Send(p.rw, NodeDataMsg, data)
 }
 
+// SendReceipts sends a batch of transaction receipts, corresponding to the ones
+// requested.
+func (p *peer) SendReceipts(receipts []*types.Receipt) error {
+	return p2p.Send(p.rw, ReceiptsMsg, receipts)
+}
+
 // RequestHashes fetches a batch of hashes from a peer, starting at from, going
 // towards the genesis block.
 func (p *peer) RequestHashes(from common.Hash) error {
@@ -205,6 +211,12 @@ func (p *peer) RequestNodeData(hashes []common.Hash) error {
 	return p2p.Send(p.rw, GetNodeDataMsg, hashes)
 }
 
+// RequestReceipts fetches a batch of transaction receipts from a remote node.
+func (p *peer) RequestReceipts(hashes []common.Hash) error {
+	glog.V(logger.Debug).Infof("%v fetching %v receipts\n", p, len(hashes))
+	return p2p.Send(p.rw, GetReceiptsMsg, hashes)
+}
+
 // Handshake executes the eth protocol handshake, negotiating version number,
 // network IDs, difficulties, head and genesis blocks.
 func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash) error {