eth: add state retrieval, test it + block tests

eth/backend.go

@@ -373,7 +373,7 @@ func New(config *Config) (*Ethereum, error) {
 
 	eth.blockProcessor = core.NewBlockProcessor(chainDb, eth.pow, eth.chainManager, eth.EventMux())
 	eth.chainManager.SetProcessor(eth.blockProcessor)
-	eth.protocolManager = NewProtocolManager(config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager)
+	eth.protocolManager = NewProtocolManager(config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager, chainDb)
 
 	eth.miner = miner.New(eth, eth.EventMux(), eth.pow)
 	eth.miner.SetGasPrice(config.GasPrice)

eth/downloader/downloader.go

@@ -47,6 +47,7 @@ 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
 
 	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

@@ -36,7 +36,7 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
-// This is the target maximum size of returned blocks or headers.
+// This is the target maximum size of returned blocks, headers or node data.
 const softResponseLimit = 2 * 1024 * 1024
 
 func errResp(code errCode, format string, v ...interface{}) error {
@@ -59,6 +59,7 @@ func (ep extProt) GetBlock(hashes []common.Hash) error { return ep.getBlocks(has
 type ProtocolManager struct {
 	txpool     txPool
 	chainman   *core.ChainManager
+	statedb    common.Database
 	downloader *downloader.Downloader
 	fetcher    *fetcher.Fetcher
 	peers      *peerSet
@@ -82,12 +83,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) *ProtocolManager {
+func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, statedb common.Database) *ProtocolManager {
 	// Create the protocol manager with the base fields
 	manager := &ProtocolManager{
 		eventMux:  mux,
 		txpool:    txpool,
 		chainman:  chainman,
+		statedb:   statedb,
 		peers:     newPeerSet(),
 		newPeerCh: make(chan *peer, 1),
 		txsyncCh:  make(chan *txsync),
@@ -276,10 +278,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 
 	case BlockHashesMsg:
 		// A batch of hashes arrived to one of our previous requests
-		msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
-
 		var hashes []common.Hash
-		if err := msgStream.Decode(&hashes); err != nil {
+		if err := msg.Decode(&hashes); err != nil {
 			break
 		}
 		// Deliver them all to the downloader for queuing
@@ -318,10 +318,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 
 	case BlocksMsg:
 		// Decode the arrived block message
-		msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
-
 		var blocks []*types.Block
-		if err := msgStream.Decode(&blocks); err != nil {
+		if err := msg.Decode(&blocks); err != nil {
 			glog.V(logger.Detail).Infoln("Decode error", err)
 			blocks = nil
 		}
@@ -361,12 +359,37 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		return p.SendBlockHeaders(headers)
 
+	case GetNodeDataMsg:
+		// 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
+			data  [][]byte
+		)
+		for bytes < softResponseLimit && len(data) < downloader.MaxStateFetch {
+			//Retrieve the hash of the next state entry
+			if err := msgStream.Decode(&hash); err == rlp.EOL {
+				break
+			} else if err != nil {
+				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 {
+				data = append(data, entry)
+				bytes += len(entry)
+			}
+		}
+		return p.SendNodeData(data)
+
 	case NewBlockHashesMsg:
 		// Retrieve and deseralize the remote new block hashes notification
-		msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
-
 		var hashes []common.Hash
-		if err := msgStream.Decode(&hashes); err != nil {
+		if err := msg.Decode(&hashes); err != nil {
 			break
 		}
 		// Mark the hashes as present at the remote node

eth/handler_test.go

@@ -1,17 +1,26 @@
 package eth
 
 import (
+	"fmt"
+	"math/big"
+	"math/rand"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth/downloader"
+	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/params"
 )
 
 // Tests that hashes can be retrieved from a remote chain by hashes in reverse
 // order.
 func TestGetBlockHashes(t *testing.T) {
-	pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil)
+	pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
 	peer, _ := newTestPeer("peer", pm, true)
 	defer peer.close()
 
@@ -52,7 +61,7 @@ func TestGetBlockHashes(t *testing.T) {
 // Tests that hashes can be retrieved from a remote chain by numbers in forward
 // order.
 func TestGetBlockHashesFromNumber(t *testing.T) {
-	pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil)
+	pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
 	peer, _ := newTestPeer("peer", pm, true)
 	defer peer.close()
 
@@ -87,3 +96,225 @@ func TestGetBlockHashesFromNumber(t *testing.T) {
 		}
 	}
 }
+
+// Tests that blocks can be retrieved from a remote chain based on their hashes.
+func TestGetBlocks(t *testing.T) {
+	pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
+	peer, _ := newTestPeer("peer", pm, true)
+	defer peer.close()
+
+	// Create a batch of tests for various scenarios
+	limit := downloader.MaxBlockFetch
+	tests := []struct {
+		random    int           // Number of blocks to fetch randomly from the chain
+		explicit  []common.Hash // Explicitly requested blocks
+		available []bool        // Availability of explicitly requested blocks
+		expected  int           // Total number of existing blocks to expect
+	}{
+		{1, nil, nil, 1},                                                       // A single random block should be retrievable
+		{10, nil, nil, 10},                                                     // Multiple random blocks should be retrievable
+		{limit, nil, nil, limit},                                               // The maximum possible blocks should be retrievable
+		{limit + 1, nil, nil, limit},                                           // No more that the possible block count should be returned
+		{0, []common.Hash{pm.chainman.Genesis().Hash()}, []bool{true}, 1},      // The genesis block should be retrievable
+		{0, []common.Hash{pm.chainman.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable
+		{0, []common.Hash{common.Hash{}}, []bool{false}, 0},                    // A non existent block should not be returned
+
+		// Existing and non-existing blocks interleaved should not cause problems
+		{0, []common.Hash{
+			common.Hash{},
+			pm.chainman.GetBlockByNumber(1).Hash(),
+			common.Hash{},
+			pm.chainman.GetBlockByNumber(10).Hash(),
+			common.Hash{},
+			pm.chainman.GetBlockByNumber(100).Hash(),
+			common.Hash{},
+		}, []bool{false, true, false, true, false, true, false}, 3},
+	}
+	// Run each of the tests and verify the results against the chain
+	for i, tt := range tests {
+		// Collect the hashes to request, and the response to expect
+		hashes, seen := []common.Hash{}, make(map[int64]bool)
+		blocks := []*types.Block{}
+
+		for j := 0; j < tt.random; j++ {
+			for {
+				num := rand.Int63n(int64(pm.chainman.CurrentBlock().NumberU64()))
+				if !seen[num] {
+					seen[num] = true
+
+					block := pm.chainman.GetBlockByNumber(uint64(num))
+					hashes = append(hashes, block.Hash())
+					if len(blocks) < tt.expected {
+						blocks = append(blocks, block)
+					}
+					break
+				}
+			}
+		}
+		for j, hash := range tt.explicit {
+			hashes = append(hashes, hash)
+			if tt.available[j] && len(blocks) < tt.expected {
+				blocks = append(blocks, pm.chainman.GetBlock(hash))
+			}
+		}
+		// Send the hash request and verify the response
+		p2p.Send(peer.app, 0x05, hashes)
+		if err := p2p.ExpectMsg(peer.app, 0x06, blocks); err != nil {
+			t.Errorf("test %d: blocks mismatch: %v", i, err)
+		}
+	}
+}
+
+// Tests that block headers can be retrieved from a remote chain based on their hashes.
+func TestGetBlockHeaders(t *testing.T) {
+	pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
+	peer, _ := newTestPeer("peer", pm, true)
+	defer peer.close()
+
+	// Create a batch of tests for various scenarios
+	limit := downloader.MaxHeaderFetch
+	tests := []struct {
+		random    int           // Number of blocks to fetch randomly from the chain
+		explicit  []common.Hash // Explicitly requested blocks
+		available []bool        // Availability of explicitly requested blocks
+		expected  int           // Total number of existing blocks to expect
+	}{
+		{1, nil, nil, 1},                                                       // A single random block should be retrievable
+		{10, nil, nil, 10},                                                     // Multiple random blocks should be retrievable
+		{limit, nil, nil, limit},                                               // The maximum possible blocks should be retrievable
+		{limit + 1, nil, nil, limit},                                           // No more that the possible block count should be returned
+		{0, []common.Hash{pm.chainman.Genesis().Hash()}, []bool{true}, 1},      // The genesis block should be retrievable
+		{0, []common.Hash{pm.chainman.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable
+		{0, []common.Hash{common.Hash{}}, []bool{false}, 0},                    // A non existent block should not be returned
+
+		// Existing and non-existing blocks interleaved should not cause problems
+		{0, []common.Hash{
+			common.Hash{},
+			pm.chainman.GetBlockByNumber(1).Hash(),
+			common.Hash{},
+			pm.chainman.GetBlockByNumber(10).Hash(),
+			common.Hash{},
+			pm.chainman.GetBlockByNumber(100).Hash(),
+			common.Hash{},
+		}, []bool{false, true, false, true, false, true, false}, 3},
+	}
+	// Run each of the tests and verify the results against the chain
+	for i, tt := range tests {
+		// Collect the hashes to request, and the response to expect
+		hashes, seen := []common.Hash{}, make(map[int64]bool)
+		headers := []*types.Header{}
+
+		for j := 0; j < tt.random; j++ {
+			for {
+				num := rand.Int63n(int64(pm.chainman.CurrentBlock().NumberU64()))
+				if !seen[num] {
+					seen[num] = true
+
+					block := pm.chainman.GetBlockByNumber(uint64(num))
+					hashes = append(hashes, block.Hash())
+					if len(headers) < tt.expected {
+						headers = append(headers, block.Header())
+					}
+					break
+				}
+			}
+		}
+		for j, hash := range tt.explicit {
+			hashes = append(hashes, hash)
+			if tt.available[j] && len(headers) < tt.expected {
+				headers = append(headers, pm.chainman.GetBlock(hash).Header())
+			}
+		}
+		// Send the hash request and verify the response
+		p2p.Send(peer.app, 0x09, hashes)
+		if err := p2p.ExpectMsg(peer.app, 0x0a, headers); err != nil {
+			t.Errorf("test %d: headers mismatch: %v", i, err)
+		}
+	}
+}
+
+// Tests that the node state database can be retrieved based on hashes.
+func TestGetNodeData(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", pm, true)
+	defer peer.close()
+
+	// Fetch for now the entire state db
+	hashes := []common.Hash{}
+	for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() {
+		hashes = append(hashes, common.BytesToHash(key))
+	}
+	p2p.Send(peer.app, 0x0b, hashes)
+	msg, err := peer.app.ReadMsg()
+	if err != nil {
+		t.Fatalf("failed to read node data response: %v", err)
+	}
+	if msg.Code != 0x0c {
+		t.Fatalf("response packet code mismatch: have %x, want %x", msg.Code, 0x0c)
+	}
+	var data [][]byte
+	if err := msg.Decode(&data); err != nil {
+		t.Fatalf("failed to decode response node data: %v", err)
+	}
+	// Verify that all hashes correspond to the requested data, and reconstruct a state tree
+	for i, want := range hashes {
+		if hash := crypto.Sha3Hash(data[i]); hash != want {
+			fmt.Errorf("data hash mismatch: have %x, want %x", hash, want)
+		}
+	}
+	statedb, _ := ethdb.NewMemDatabase()
+	for i := 0; i < len(data); i++ {
+		statedb.Put(hashes[i].Bytes(), data[i])
+	}
+	accounts := []common.Address{testBankAddress, acc1Addr, acc2Addr}
+	for i := uint64(0); i <= pm.chainman.CurrentBlock().NumberU64(); i++ {
+		trie := state.New(pm.chainman.GetBlockByNumber(i).Root(), statedb)
+
+		for j, acc := range accounts {
+			bw := pm.chainman.State().GetBalance(acc)
+			bh := trie.GetBalance(acc)
+
+			if (bw != nil && bh == nil) || (bw == nil && bh != nil) {
+				t.Errorf("test %d, account %d: balance mismatch: have %v, want %v", i, j, bh, bw)
+			}
+			if bw != nil && bh != nil && bw.Cmp(bw) != 0 {
+				t.Errorf("test %d, account %d: balance mismatch: have %v, want %v", i, j, bh, bw)
+			}
+		}
+	}
+}

eth/helper_test.go

@@ -19,22 +19,29 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
+var (
+	testBankKey, _  = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+	testBankAddress = crypto.PubkeyToAddress(testBankKey.PublicKey)
+	testBankFunds   = big.NewInt(1000000)
+)
+
 // newTestProtocolManager creates a new protocol manager for testing purposes,
 // with the given number of blocks already known, and potential notification
 // channels for different events.
-func newTestProtocolManager(blocks int, newtx chan<- []*types.Transaction) *ProtocolManager {
+func newTestProtocolManager(blocks int, generator func(int, *core.BlockGen), newtx chan<- []*types.Transaction) *ProtocolManager {
 	var (
-		emux        = new(event.TypeMux)
+		evmux       = new(event.TypeMux)
 		pow         = new(core.FakePow)
 		db, _       = ethdb.NewMemDatabase()
-		genesis     = core.WriteGenesisBlockForTesting(db, common.Address{}, big.NewInt(0))
+		genesis     = core.WriteGenesisBlockForTesting(db, testBankAddress, testBankFunds)
-		chainman, _ = core.NewChainManager(db, pow, emux)
+		chainman, _ = core.NewChainManager(db, pow, evmux)
-		blockproc   = core.NewBlockProcessor(db, pow, chainman, emux)
+		blockproc   = core.NewBlockProcessor(db, pow, chainman, evmux)
 	)
 	chainman.SetProcessor(blockproc)
-	chainman.InsertChain(core.GenerateChain(genesis, db, blocks, nil))
+	if _, err := chainman.InsertChain(core.GenerateChain(genesis, db, blocks, generator)); err != nil {
-
+		panic(err)
-	pm := NewProtocolManager(NetworkId, emux, &testTxPool{added: newtx}, pow, chainman)
+	}
+	pm := NewProtocolManager(NetworkId, evmux, &testTxPool{added: newtx}, pow, chainman, db)
 	pm.Start()
 	return pm
 }

eth/protocol_test.go

@@ -37,7 +37,7 @@ func init() {
 var testAccount = crypto.NewKey(rand.Reader)
 
 func TestStatusMsgErrors(t *testing.T) {
-	pm := newTestProtocolManager(0, nil)
+	pm := newTestProtocolManager(0, nil, nil)
 	td, currentBlock, genesis := pm.chainman.Status()
 	defer pm.Stop()
 
@@ -87,7 +87,7 @@ func TestStatusMsgErrors(t *testing.T) {
 // This test checks that received transactions are added to the local pool.
 func TestRecvTransactions(t *testing.T) {
 	txAdded := make(chan []*types.Transaction)
-	pm := newTestProtocolManager(0, txAdded)
+	pm := newTestProtocolManager(0, nil, txAdded)
 	p, _ := newTestPeer("peer", pm, true)
 	defer pm.Stop()
 	defer p.close()
@@ -110,7 +110,7 @@ func TestRecvTransactions(t *testing.T) {
 
 // This test checks that pending transactions are sent.
 func TestSendTransactions(t *testing.T) {
-	pm := newTestProtocolManager(0, nil)
+	pm := newTestProtocolManager(0, nil, nil)
 	defer pm.Stop()
 
 	// Fill the pool with big transactions.

ethdb/memory_database.go

@@ -49,6 +49,14 @@ func (db *MemDatabase) Get(key []byte) ([]byte, error) {
 	return db.db[string(key)], nil
 }
 
+func (db *MemDatabase) Keys() [][]byte {
+	keys := [][]byte{}
+	for key, _ := range db.db {
+		keys = append(keys, []byte(key))
+	}
+	return keys
+}
+
 /*
 func (db *MemDatabase) GetKeys() []*common.Key {
 	data, _ := db.Get([]byte("KeyRing"))