eth, eth/downloader: handle header requests, table driven proto tests

cmd/geth/main.go

@@ -283,6 +283,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso
 		utils.DataDirFlag,
 		utils.BlockchainVersionFlag,
 		utils.OlympicFlag,
+		utils.EthVersionFlag,
 		utils.CacheFlag,
 		utils.JSpathFlag,
 		utils.ListenPortFlag,
@@ -333,6 +334,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso
 	app.Before = func(ctx *cli.Context) error {
 		utils.SetupLogger(ctx)
 		utils.SetupVM(ctx)
+		utils.SetupEth(ctx)
 		if ctx.GlobalBool(utils.PProfEanbledFlag.Name) {
 			utils.StartPProf(ctx)
 		}

cmd/utils/flags.go

@@ -138,6 +138,11 @@ var (
 		Name:  "olympic",
 		Usage: "Use olympic style protocol",
 	}
+	EthVersionFlag = cli.IntFlag{
+		Name:  "eth",
+		Value: 61,
+		Usage: "Highest eth protocol to advertise (temporary, dev option)",
+	}
 
 	// miner settings
 	MinerThreadsFlag = cli.IntFlag{
@@ -459,6 +464,18 @@ func SetupVM(ctx *cli.Context) {
 	vm.SetJITCacheSize(ctx.GlobalInt(VMJitCacheFlag.Name))
 }
 
+// SetupEth configures the eth packages global settings
+func SetupEth(ctx *cli.Context) {
+	version := ctx.GlobalInt(EthVersionFlag.Name)
+	for len(eth.ProtocolVersions) > 0 && eth.ProtocolVersions[0] > uint(version) {
+		eth.ProtocolVersions = eth.ProtocolVersions[1:]
+		eth.ProtocolLengths = eth.ProtocolLengths[1:]
+	}
+	if len(eth.ProtocolVersions) == 0 {
+		Fatalf("No valid eth protocols remaining")
+	}
+}
+
 // MakeChain creates a chain manager from set command line flags.
 func MakeChain(ctx *cli.Context) (chain *core.ChainManager, chainDb common.Database) {
 	datadir := ctx.GlobalString(DataDirFlag.Name)

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

@@ -39,13 +39,15 @@ import (
 const (
 	eth60 = 60 // Constant to check for old protocol support
 	eth61 = 61 // Constant to check for new protocol support
-	eth62 = 62 // Constant to check for experimental protocol support
 )
 
 var (
 	MinHashFetch     = 512 // Minimum amount of hashes to not consider a peer stalling
 	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 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
@@ -330,7 +332,7 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
 		if err = d.fetchBlocks60(); err != nil {
 			return err
 		}
-	case eth61, eth62:
+	case eth61:
 		// New eth/61, use forward, concurrent hash and block retrieval algorithm
 		number, err := d.findAncestor(p)
 		if err != nil {

eth/fetcher/fetcher.go

@@ -70,6 +70,7 @@ type peerDropFn func(id string)
 // network.
 type announce struct {
 	hash   common.Hash // Hash of the block being announced
+	number uint64      // Number of the block being announced (0 = unknown | old protocol)
 	time   time.Time   // Timestamp of the announcement
 
 	origin string           // Identifier of the peer originating the notification
@@ -152,9 +153,10 @@ func (f *Fetcher) Stop() {
 
 // Notify announces the fetcher of the potential availability of a new block in
 // the network.
-func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher blockRequesterFn) error {
+func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time, fetcher blockRequesterFn) error {
 	block := &announce{
 		hash:   hash,
+		number: number,
 		time:   time,
 		origin: peer,
 		fetch:  fetcher,

eth/fetcher/fetcher_test.go

@@ -194,7 +194,7 @@ func TestSequentialAnnouncements(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	for i := len(hashes) - 2; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+		tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 		verifyImportEvent(t, imported)
 	}
 	verifyImportDone(t, imported)
@@ -221,9 +221,9 @@ func TestConcurrentAnnouncements(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	for i := len(hashes) - 2; i >= 0; i-- {
-		tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper)
-		tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
-		tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
+		tester.fetcher.Notify("first", hashes[i], 0, time.Now().Add(-arriveTimeout), wrapper)
+		tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
+		tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
 
 		verifyImportEvent(t, imported)
 	}
@@ -252,7 +252,7 @@ func TestOverlappingAnnouncements(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	for i := len(hashes) - 2; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+		tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 		select {
 		case <-fetching:
 		case <-time.After(time.Second):
@@ -286,7 +286,7 @@ func TestPendingDeduplication(t *testing.T) {
 	}
 	// Announce the same block many times until it's fetched (wait for any pending ops)
 	for tester.getBlock(hashes[0]) == nil {
-		tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper)
+		tester.fetcher.Notify("repeater", hashes[0], 0, time.Now().Add(-arriveTimeout), wrapper)
 		time.Sleep(time.Millisecond)
 	}
 	time.Sleep(delay)
@@ -317,12 +317,12 @@ func TestRandomArrivalImport(t *testing.T) {
 
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
-			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+			tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 			time.Sleep(time.Millisecond)
 		}
 	}
 	// Finally announce the skipped entry and check full import
-	tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher)
+	tester.fetcher.Notify("valid", hashes[skip], 0, time.Now().Add(-arriveTimeout), fetcher)
 	verifyImportCount(t, imported, len(hashes)-1)
 }
 
@@ -343,7 +343,7 @@ func TestQueueGapFill(t *testing.T) {
 
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
-			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+			tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 			time.Sleep(time.Millisecond)
 		}
 	}
@@ -374,7 +374,7 @@ func TestImportDeduplication(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	// Announce the duplicating block, wait for retrieval, and also propagate directly
-	tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher)
+	tester.fetcher.Notify("valid", hashes[0], 0, time.Now().Add(-arriveTimeout), fetcher)
 	<-fetching
 
 	tester.fetcher.Enqueue("valid", blocks[hashes[0]])
@@ -437,9 +437,9 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
 	// Feed the tester a huge hashset from the attacker, and a limited from the valid peer
 	for i := 0; i < len(attack); i++ {
 		if i < maxQueueDist {
-			tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], time.Now(), valid)
+			tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], 0, time.Now(), valid)
 		}
-		tester.fetcher.Notify("attacker", attack[i], time.Now(), attacker)
+		tester.fetcher.Notify("attacker", attack[i], 0, time.Now(), attacker)
 	}
 	if len(tester.fetcher.announced) != hashLimit+maxQueueDist {
 		t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist)
@@ -449,7 +449,7 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
 
 	// Feed the remaining valid hashes to ensure DOS protection state remains clean
 	for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid)
+		tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), valid)
 		verifyImportEvent(t, imported)
 	}
 	verifyImportDone(t, imported)

eth/helper_test.go

+// This file contains some shares testing functionality, common to  multiple
+// different files and modules being tested.
+
+package eth
+
+import (
+	"crypto/rand"
+	"math/big"
+	"sync"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/p2p"
+	"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, generator func(int, *core.BlockGen), newtx chan<- []*types.Transaction) *ProtocolManager {
+	var (
+		evmux       = new(event.TypeMux)
+		pow         = new(core.FakePow)
+		db, _       = ethdb.NewMemDatabase()
+		genesis     = core.WriteGenesisBlockForTesting(db, testBankAddress, testBankFunds)
+		chainman, _ = core.NewChainManager(db, pow, evmux)
+		blockproc   = core.NewBlockProcessor(db, pow, chainman, evmux)
+	)
+	chainman.SetProcessor(blockproc)
+	if _, err := chainman.InsertChain(core.GenerateChain(genesis, db, blocks, generator)); err != nil {
+		panic(err)
+	}
+	pm := NewProtocolManager(NetworkId, evmux, &testTxPool{added: newtx}, pow, chainman, db)
+	pm.Start()
+	return pm
+}
+
+// testTxPool is a fake, helper transaction pool for testing purposes
+type testTxPool struct {
+	pool  []*types.Transaction        // Collection of all transactions
+	added chan<- []*types.Transaction // Notification channel for new transactions
+
+	lock sync.RWMutex // Protects the transaction pool
+}
+
+// AddTransactions appends a batch of transactions to the pool, and notifies any
+// listeners if the addition channel is non nil
+func (p *testTxPool) AddTransactions(txs []*types.Transaction) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	p.pool = append(p.pool, txs...)
+	if p.added != nil {
+		p.added <- txs
+	}
+}
+
+// GetTransactions returns all the transactions known to the pool
+func (p *testTxPool) GetTransactions() types.Transactions {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	txs := make([]*types.Transaction, len(p.pool))
+	copy(txs, p.pool)
+
+	return txs
+}
+
+// newTestTransaction create a new dummy transaction.
+func newTestTransaction(from *crypto.Key, nonce uint64, datasize int) *types.Transaction {
+	tx := types.NewTransaction(nonce, common.Address{}, big.NewInt(0), big.NewInt(100000), big.NewInt(0), make([]byte, datasize))
+	tx, _ = tx.SignECDSA(from.PrivateKey)
+
+	return tx
+}
+
+// testPeer is a simulated peer to allow testing direct network calls.
+type testPeer struct {
+	net p2p.MsgReadWriter // Network layer reader/writer to simulate remote messaging
+	app *p2p.MsgPipeRW    // Application layer reader/writer to simulate the local side
+	*peer
+}
+
+// newTestPeer creates a new peer registered at the given protocol manager.
+func newTestPeer(name string, version int, pm *ProtocolManager, shake bool) (*testPeer, <-chan error) {
+	// Create a message pipe to communicate through
+	app, net := p2p.MsgPipe()
+
+	// Generate a random id and create the peer
+	var id discover.NodeID
+	rand.Read(id[:])
+
+	peer := pm.newPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
+
+	// Start the peer on a new thread
+	errc := make(chan error, 1)
+	go func() {
+		pm.newPeerCh <- peer
+		errc <- pm.handle(peer)
+	}()
+	tp := &testPeer{
+		app:  app,
+		net:  net,
+		peer: peer,
+	}
+	// Execute any implicitly requested handshakes and return
+	if shake {
+		td, head, genesis := pm.chainman.Status()
+		tp.handshake(nil, td, head, genesis)
+	}
+	return tp, errc
+}
+
+// handshake simulates a trivial handshake that expects the same state from the
+// remote side as we are simulating locally.
+func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, genesis common.Hash) {
+	msg := &statusData{
+		ProtocolVersion: uint32(p.version),
+		NetworkId:       uint32(NetworkId),
+		TD:              td,
+		CurrentBlock:    head,
+		GenesisBlock:    genesis,
+	}
+	if err := p2p.ExpectMsg(p.app, StatusMsg, msg); err != nil {
+		t.Fatalf("status recv: %v", err)
+	}
+	if err := p2p.Send(p.app, StatusMsg, msg); err != nil {
+		t.Fatalf("status send: %v", err)
+	}
+}
+
+// close terminates the local side of the peer, notifying the remote protocol
+// manager of termination.
+func (p *testPeer) close() {
+	p.app.Close()
+}

eth/metrics.go

@@ -46,10 +46,18 @@ var (
 	reqHeaderInTrafficMeter   = metrics.NewMeter("eth/req/header/in/traffic")
 	reqHeaderOutPacketsMeter  = metrics.NewMeter("eth/req/header/out/packets")
 	reqHeaderOutTrafficMeter  = metrics.NewMeter("eth/req/header/out/traffic")
+	reqBodyInPacketsMeter     = metrics.NewMeter("eth/req/body/in/packets")
+	reqBodyInTrafficMeter     = metrics.NewMeter("eth/req/body/in/traffic")
+	reqBodyOutPacketsMeter    = metrics.NewMeter("eth/req/body/out/packets")
+	reqBodyOutTrafficMeter    = metrics.NewMeter("eth/req/body/out/traffic")
 	reqStateInPacketsMeter    = metrics.NewMeter("eth/req/state/in/packets")
 	reqStateInTrafficMeter    = metrics.NewMeter("eth/req/state/in/traffic")
 	reqStateOutPacketsMeter   = metrics.NewMeter("eth/req/state/out/packets")
 	reqStateOutTrafficMeter   = metrics.NewMeter("eth/req/state/out/traffic")
+	reqReceiptInPacketsMeter  = metrics.NewMeter("eth/req/receipt/in/packets")
+	reqReceiptInTrafficMeter  = metrics.NewMeter("eth/req/receipt/in/traffic")
+	reqReceiptOutPacketsMeter = metrics.NewMeter("eth/req/receipt/out/packets")
+	reqReceiptOutTrafficMeter = metrics.NewMeter("eth/req/receipt/out/traffic")
 	miscInPacketsMeter        = metrics.NewMeter("eth/misc/in/packets")
 	miscInTrafficMeter        = metrics.NewMeter("eth/misc/in/traffic")
 	miscOutPacketsMeter       = metrics.NewMeter("eth/misc/out/packets")
@@ -59,7 +67,8 @@ var (
 // meteredMsgReadWriter is a wrapper around a p2p.MsgReadWriter, capable of
 // accumulating the above defined metrics based on the data stream contents.
 type meteredMsgReadWriter struct {
-	p2p.MsgReadWriter
+	p2p.MsgReadWriter     // Wrapped message stream to meter
+	version           int // Protocol version to select correct meters
 }
 
 // newMeteredMsgWriter wraps a p2p MsgReadWriter with metering support. If the
@@ -68,7 +77,13 @@ func newMeteredMsgWriter(rw p2p.MsgReadWriter) p2p.MsgReadWriter {
 	if !metrics.Enabled {
 		return rw
 	}
-	return &meteredMsgReadWriter{rw}
+	return &meteredMsgReadWriter{MsgReadWriter: rw}
+}
+
+// Init sets the protocol version used by the stream to know which meters to
+// increment in case of overlapping message ids between protocol versions.
+func (rw *meteredMsgReadWriter) Init(version int) {
+	rw.version = version
 }
 
 func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
@@ -79,20 +94,27 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
 	}
 	// Account for the data traffic
 	packets, traffic := miscInPacketsMeter, miscInTrafficMeter
-	switch msg.Code {
-	case BlockHashesMsg:
+	switch {
+	case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlockHashesMsg:
 		packets, traffic = reqHashInPacketsMeter, reqHashInTrafficMeter
-	case BlocksMsg:
+	case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlocksMsg:
 		packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter
-	case BlockHeadersMsg:
-		packets, traffic = reqHeaderInPacketsMeter, reqHeaderInTrafficMeter
-	case NodeDataMsg:
+
+	case rw.version == eth62 && msg.Code == BlockHeadersMsg:
+		packets, traffic = reqBlockInPacketsMeter, reqBlockInTrafficMeter
+	case rw.version == eth62 && msg.Code == BlockBodiesMsg:
+		packets, traffic = reqBodyInPacketsMeter, reqBodyInTrafficMeter
+
+	case rw.version == eth63 && msg.Code == NodeDataMsg:
 		packets, traffic = reqStateInPacketsMeter, reqStateInTrafficMeter
-	case NewBlockHashesMsg:
+	case rw.version == eth63 && msg.Code == ReceiptsMsg:
+		packets, traffic = reqReceiptInPacketsMeter, reqReceiptInTrafficMeter
+
+	case msg.Code == NewBlockHashesMsg:
 		packets, traffic = propHashInPacketsMeter, propHashInTrafficMeter
-	case NewBlockMsg:
+	case msg.Code == NewBlockMsg:
 		packets, traffic = propBlockInPacketsMeter, propBlockInTrafficMeter
-	case TxMsg:
+	case msg.Code == TxMsg:
 		packets, traffic = propTxnInPacketsMeter, propTxnInTrafficMeter
 	}
 	packets.Mark(1)
@@ -104,20 +126,27 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
 func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error {
 	// Account for the data traffic
 	packets, traffic := miscOutPacketsMeter, miscOutTrafficMeter
-	switch msg.Code {
-	case BlockHashesMsg:
+	switch {
+	case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlockHashesMsg:
 		packets, traffic = reqHashOutPacketsMeter, reqHashOutTrafficMeter
-	case BlocksMsg:
+	case (rw.version == eth60 || rw.version == eth61) && msg.Code == BlocksMsg:
 		packets, traffic = reqBlockOutPacketsMeter, reqBlockOutTrafficMeter
-	case BlockHeadersMsg:
+
+	case rw.version == eth62 && msg.Code == BlockHeadersMsg:
 		packets, traffic = reqHeaderOutPacketsMeter, reqHeaderOutTrafficMeter
-	case NodeDataMsg:
+	case rw.version == eth62 && msg.Code == BlockBodiesMsg:
+		packets, traffic = reqBodyOutPacketsMeter, reqBodyOutTrafficMeter
+
+	case rw.version == eth63 && msg.Code == NodeDataMsg:
 		packets, traffic = reqStateOutPacketsMeter, reqStateOutTrafficMeter
-	case NewBlockHashesMsg:
+	case rw.version == eth63 && msg.Code == ReceiptsMsg:
+		packets, traffic = reqReceiptOutPacketsMeter, reqReceiptOutTrafficMeter
+
+	case msg.Code == NewBlockHashesMsg:
 		packets, traffic = propHashOutPacketsMeter, propHashOutTrafficMeter
-	case NewBlockMsg:
+	case msg.Code == NewBlockMsg:
 		packets, traffic = propBlockOutPacketsMeter, propBlockOutTrafficMeter
-	case TxMsg:
+	case msg.Code == TxMsg:
 		packets, traffic = propTxnOutPacketsMeter, propTxnOutTrafficMeter
 	}
 	packets.Mark(1)

eth/peer.go

@@ -165,12 +165,23 @@ func (p *peer) SendBlockHeaders(headers []*types.Header) error {
 	return p2p.Send(p.rw, BlockHeadersMsg, headers)
 }
 
+// SendBlockBodies sends a batch of block contents to the remote peer.
+func (p *peer) SendBlockBodies(bodies []*blockBody) error {
+	return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies))
+}
+
 // SendNodeData sends a batch of arbitrary internal data, corresponding to the
 // hashes requested.
 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 +216,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 {

eth/protocol.go

@@ -17,17 +17,29 @@
 package eth
 
 import (
+	"fmt"
+	"io"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+// Constants to match up protocol versions and messages
+const (
+	eth60 = 60
+	eth61 = 61
+	eth62 = 62
+	eth63 = 63
+	eth64 = 64
 )
 
 // Supported versions of the eth protocol (first is primary).
-var ProtocolVersions = []uint{62, 61, 60}
+var ProtocolVersions = []uint{eth64, eth63, eth62, eth61, eth60}
 
 // Number of implemented message corresponding to different protocol versions.
-var ProtocolLengths = []uint64{13, 9, 8}
+var ProtocolLengths = []uint64{15, 12, 8, 9, 8}
 
 const (
 	NetworkId          = 1
@@ -37,23 +49,38 @@ const (
 // eth protocol message codes
 const (
 	// Protocol messages belonging to eth/60
-	StatusMsg = iota
-	NewBlockHashesMsg
-	TxMsg
-	GetBlockHashesMsg
-	BlockHashesMsg
-	GetBlocksMsg
-	BlocksMsg
-	NewBlockMsg
-
-	// Protocol messages belonging to eth/61
-	GetBlockHashesFromNumberMsg
-
-	// Protocol messages belonging to eth/62
-	GetBlockHeadersMsg
-	BlockHeadersMsg
-	GetNodeDataMsg
-	NodeDataMsg
+	StatusMsg         = 0x00
+	NewBlockHashesMsg = 0x01
+	TxMsg             = 0x02
+	GetBlockHashesMsg = 0x03
+	BlockHashesMsg    = 0x04
+	GetBlocksMsg      = 0x05
+	BlocksMsg         = 0x06
+	NewBlockMsg       = 0x07
+
+	// Protocol messages belonging to eth/61 (extension of eth/60)
+	GetBlockHashesFromNumberMsg = 0x08
+
+	// Protocol messages belonging to eth/62 (new protocol from scratch)
+	// StatusMsg          = 0x00 (uncomment after eth/61 deprecation)
+	// NewBlockHashesMsg  = 0x01 (uncomment after eth/61 deprecation)
+	// TxMsg              = 0x02 (uncomment after eth/61 deprecation)
+	GetBlockHeadersMsg = 0x03
+	BlockHeadersMsg    = 0x04
+	GetBlockBodiesMsg  = 0x05
+	BlockBodiesMsg     = 0x06
+	// 	NewBlockMsg       = 0x07 (uncomment after eth/61 deprecation)
+
+	// Protocol messages belonging to eth/63
+	GetNodeDataMsg = 0x0d
+	NodeDataMsg    = 0x0e
+	GetReceiptsMsg = 0x0f
+	ReceiptsMsg    = 0x10
+
+	// Protocol messages belonging to eth/64
+	GetAcctProofMsg     = 0x11
+	GetStorageDataProof = 0x12
+	Proof               = 0x13
 )
 
 type errCode int
@@ -111,6 +138,12 @@ type statusData struct {
 	GenesisBlock    common.Hash
 }
 
+// newBlockHashesData is the network packet for the block announcements.
+type newBlockHashesData []struct {
+	Hash   common.Hash // Hash of one particular block being announced
+	Number uint64      // Number of one particular block being announced
+}
+
 // getBlockHashesData is the network packet for the hash based hash retrieval.
 type getBlockHashesData struct {
 	Hash   common.Hash
@@ -124,12 +157,65 @@ type getBlockHashesFromNumberData struct {
 	Amount uint64
 }
 
+// getBlockHeadersData represents a block header query.
+type getBlockHeadersData struct {
+	Origin  hashOrNumber // Block from which to retrieve headers
+	Amount  uint64       // Maximum number of headers to retrieve
+	Skip    uint64       // Blocks to skip between consecutive headers
+	Reverse bool         // Query direction (false = rising towards latest, true = falling towards genesis)
+}
+
+// hashOrNumber is a combined field for specifying an origin block.
+type hashOrNumber struct {
+	Hash   common.Hash // Block hash from which to retrieve headers (excludes Number)
+	Number uint64      // Block hash from which to retrieve headers (excludes Hash)
+}
+
+// EncodeRLP is a specialized encoder for hashOrNumber to encode only one of the
+// two contained union fields.
+func (hn *hashOrNumber) EncodeRLP(w io.Writer) error {
+	if hn.Hash == (common.Hash{}) {
+		return rlp.Encode(w, hn.Number)
+	}
+	if hn.Number != 0 {
+		return fmt.Errorf("both origin hash (%x) and number (%d) provided", hn.Hash, hn.Number)
+	}
+	return rlp.Encode(w, hn.Hash)
+}
+
+// DecodeRLP is a specialized decoder for hashOrNumber to decode the contents
+// into either a block hash or a block number.
+func (hn *hashOrNumber) DecodeRLP(s *rlp.Stream) error {
+	_, size, _ := s.Kind()
+	origin, err := s.Raw()
+	if err == nil {
+		switch {
+		case size == 32:
+			err = rlp.DecodeBytes(origin, &hn.Hash)
+		case size <= 8:
+			err = rlp.DecodeBytes(origin, &hn.Number)
+		default:
+			err = fmt.Errorf("invalid input size %d for origin", size)
+		}
+	}
+	return err
+}
+
 // newBlockData is the network packet for the block propagation message.
 type newBlockData struct {
 	Block *types.Block
 	TD    *big.Int
 }
 
+// blockBody represents the data content of a single block.
+type blockBody struct {
+	Transactions []*types.Transaction // Transactions contained within a block
+	Uncles       []*types.Header      // Uncles contained within a block
+}
+
+// blockBodiesData is the network packet for block content distribution.
+type blockBodiesData []*blockBody
+
 // nodeDataData is the network response packet for a node data retrieval.
 type nodeDataData []struct {
 	Value []byte

eth/protocol_test.go

@@ -18,19 +18,16 @@ package eth
 
 import (
 	"crypto/rand"
-	"math/big"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/ethdb"
-	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/p2p"
-	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/rlp"
 )
 
 func init() {
@@ -40,8 +37,15 @@ func init() {
 
 var testAccount = crypto.NewKey(rand.Reader)
 
-func TestStatusMsgErrors(t *testing.T) {
-	pm := newProtocolManagerForTesting(nil)
+// Tests that handshake failures are detected and reported correctly.
+func TestStatusMsgErrors60(t *testing.T) { testStatusMsgErrors(t, 60) }
+func TestStatusMsgErrors61(t *testing.T) { testStatusMsgErrors(t, 61) }
+func TestStatusMsgErrors62(t *testing.T) { testStatusMsgErrors(t, 62) }
+func TestStatusMsgErrors63(t *testing.T) { testStatusMsgErrors(t, 63) }
+func TestStatusMsgErrors64(t *testing.T) { testStatusMsgErrors(t, 64) }
+
+func testStatusMsgErrors(t *testing.T, protocol int) {
+	pm := newTestProtocolManager(0, nil, nil)
 	td, currentBlock, genesis := pm.chainman.Status()
 	defer pm.Stop()
 
@@ -56,23 +60,23 @@ func TestStatusMsgErrors(t *testing.T) {
 		},
 		{
 			code: StatusMsg, data: statusData{10, NetworkId, td, currentBlock, genesis},
-			wantError: errResp(ErrProtocolVersionMismatch, "10 (!= 0)"),
+			wantError: errResp(ErrProtocolVersionMismatch, "10 (!= %d)", protocol),
 		},
 		{
-			code: StatusMsg, data: statusData{uint32(ProtocolVersions[0]), 999, td, currentBlock, genesis},
+			code: StatusMsg, data: statusData{uint32(protocol), 999, td, currentBlock, genesis},
 			wantError: errResp(ErrNetworkIdMismatch, "999 (!= 1)"),
 		},
 		{
-			code: StatusMsg, data: statusData{uint32(ProtocolVersions[0]), NetworkId, td, currentBlock, common.Hash{3}},
+			code: StatusMsg, data: statusData{uint32(protocol), NetworkId, td, currentBlock, common.Hash{3}},
 			wantError: errResp(ErrGenesisBlockMismatch, "0300000000000000000000000000000000000000000000000000000000000000 (!= %x)", genesis),
 		},
 	}
 
 	for i, test := range tests {
-		p, errc := newTestPeer(pm)
+		p, errc := newTestPeer("peer", protocol, pm, false)
 		// The send call might hang until reset because
 		// the protocol might not read the payload.
-		go p2p.Send(p, test.code, test.data)
+		go p2p.Send(p.app, test.code, test.data)
 
 		select {
 		case err := <-errc:
@@ -89,16 +93,21 @@ func TestStatusMsgErrors(t *testing.T) {
 }
 
 // This test checks that received transactions are added to the local pool.
-func TestRecvTransactions(t *testing.T) {
+func TestRecvTransactions60(t *testing.T) { testRecvTransactions(t, 60) }
+func TestRecvTransactions61(t *testing.T) { testRecvTransactions(t, 61) }
+func TestRecvTransactions62(t *testing.T) { testRecvTransactions(t, 62) }
+func TestRecvTransactions63(t *testing.T) { testRecvTransactions(t, 63) }
+func TestRecvTransactions64(t *testing.T) { testRecvTransactions(t, 64) }
+
+func testRecvTransactions(t *testing.T, protocol int) {
 	txAdded := make(chan []*types.Transaction)
-	pm := newProtocolManagerForTesting(txAdded)
-	p, _ := newTestPeer(pm)
+	pm := newTestProtocolManager(0, nil, txAdded)
+	p, _ := newTestPeer("peer", protocol, pm, true)
 	defer pm.Stop()
 	defer p.close()
-	p.handshake(t)
 
-	tx := newtx(testAccount, 0, 0)
-	if err := p2p.Send(p, TxMsg, []interface{}{tx}); err != nil {
+	tx := newTestTransaction(testAccount, 0, 0)
+	if err := p2p.Send(p.app, TxMsg, []interface{}{tx}); err != nil {
 		t.Fatalf("send error: %v", err)
 	}
 	select {
@@ -114,15 +123,21 @@ func TestRecvTransactions(t *testing.T) {
 }
 
 // This test checks that pending transactions are sent.
-func TestSendTransactions(t *testing.T) {
-	pm := newProtocolManagerForTesting(nil)
+func TestSendTransactions60(t *testing.T) { testSendTransactions(t, 60) }
+func TestSendTransactions61(t *testing.T) { testSendTransactions(t, 61) }
+func TestSendTransactions62(t *testing.T) { testSendTransactions(t, 62) }
+func TestSendTransactions63(t *testing.T) { testSendTransactions(t, 63) }
+func TestSendTransactions64(t *testing.T) { testSendTransactions(t, 64) }
+
+func testSendTransactions(t *testing.T, protocol int) {
+	pm := newTestProtocolManager(0, nil, nil)
 	defer pm.Stop()
 
 	// Fill the pool with big transactions.
 	const txsize = txsyncPackSize / 10
 	alltxs := make([]*types.Transaction, 100)
 	for nonce := range alltxs {
-		alltxs[nonce] = newtx(testAccount, uint64(nonce), txsize)
+		alltxs[nonce] = newTestTransaction(testAccount, uint64(nonce), txsize)
 	}
 	pm.txpool.AddTransactions(alltxs)
 
@@ -137,7 +152,7 @@ func TestSendTransactions(t *testing.T) {
 		}
 		for n := 0; n < len(alltxs) && !t.Failed(); {
 			var txs []*types.Transaction
-			msg, err := p.ReadMsg()
+			msg, err := p.app.ReadMsg()
 			if err != nil {
 				t.Errorf("%v: read error: %v", p.Peer, err)
 			} else if msg.Code != TxMsg {
@@ -161,97 +176,53 @@ func TestSendTransactions(t *testing.T) {
 		}
 	}
 	for i := 0; i < 3; i++ {
-		p, _ := newTestPeer(pm)
-		p.handshake(t)
+		p, _ := newTestPeer(fmt.Sprintf("peer #%d", i), protocol, pm, true)
 		wg.Add(1)
 		go checktxs(p)
 	}
 	wg.Wait()
 }
 
-// testPeer wraps all peer-related data for tests.
-type testPeer struct {
-	p2p.MsgReadWriter                // writing to the test peer feeds the protocol
-	pipe              *p2p.MsgPipeRW // the protocol read/writes on this end
-	pm                *ProtocolManager
-	*peer
-}
-
-func newProtocolManagerForTesting(txAdded chan<- []*types.Transaction) *ProtocolManager {
-	db, _ := ethdb.NewMemDatabase()
-	core.WriteTestNetGenesisBlock(db, 0)
-	var (
-		em       = new(event.TypeMux)
-		chain, _ = core.NewChainManager(db, core.FakePow{}, em)
-		txpool   = &fakeTxPool{added: txAdded}
-		pm       = NewProtocolManager(NetworkId, em, txpool, core.FakePow{}, chain)
-	)
-	pm.Start()
-	return pm
-}
+// Tests that the custom union field encoder and decoder works correctly.
+func TestGetBlockHeadersDataEncodeDecode(t *testing.T) {
+	// Create a "random" hash for testing
+	var hash common.Hash
+	for i, _ := range hash {
+		hash[i] = byte(i)
+	}
+	// Assemble some table driven tests
+	tests := []struct {
+		packet *getBlockHeadersData
+		fail   bool
+	}{
+		// Providing the origin as either a hash or a number should both work
+		{fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Number: 314}}},
+		{fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}}},
 
-func newTestPeer(pm *ProtocolManager) (*testPeer, <-chan error) {
-	var id discover.NodeID
-	rand.Read(id[:])
-	rw1, rw2 := p2p.MsgPipe()
-	peer := pm.newPeer(pm.protVer, pm.netId, p2p.NewPeer(id, "test peer", nil), rw2)
-	errc := make(chan error, 1)
-	go func() {
-		pm.newPeerCh <- peer
-		errc <- pm.handle(peer)
-	}()
-	return &testPeer{rw1, rw2, pm, peer}, errc
-}
+		// Providing arbitrary query field should also work
+		{fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Number: 314}, Amount: 314, Skip: 1, Reverse: true}},
+		{fail: false, packet: &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: 314, Skip: 1, Reverse: true}},
 
-func (p *testPeer) handshake(t *testing.T) {
-	td, currentBlock, genesis := p.pm.chainman.Status()
-	msg := &statusData{
-		ProtocolVersion: uint32(p.pm.protVer),
-		NetworkId:       uint32(p.pm.netId),
-		TD:              td,
-		CurrentBlock:    currentBlock,
-		GenesisBlock:    genesis,
+		// Providing both the origin hash and origin number must fail
+		{fail: true, packet: &getBlockHeadersData{Origin: hashOrNumber{Hash: hash, Number: 314}}},
 	}
-	if err := p2p.ExpectMsg(p, StatusMsg, msg); err != nil {
-		t.Fatalf("status recv: %v", err)
+	// Iterate over each of the tests and try to encode and then decode
+	for i, tt := range tests {
+		bytes, err := rlp.EncodeToBytes(tt.packet)
+		if err != nil && !tt.fail {
+			t.Fatalf("test %d: failed to encode packet: %v", i, err)
+		} else if err == nil && tt.fail {
+			t.Fatalf("test %d: encode should have failed", i)
+		}
+		if !tt.fail {
+			packet := new(getBlockHeadersData)
+			if err := rlp.DecodeBytes(bytes, packet); err != nil {
+				t.Fatalf("test %d: failed to decode packet: %v", i, err)
+			}
+			if packet.Origin.Hash != tt.packet.Origin.Hash || packet.Origin.Number != tt.packet.Origin.Number || packet.Amount != tt.packet.Amount ||
+				packet.Skip != tt.packet.Skip || packet.Reverse != tt.packet.Reverse {
+				t.Fatalf("test %d: encode decode mismatch: have %+v, want %+v", i, packet, tt.packet)
 			}
-	if err := p2p.Send(p, StatusMsg, msg); err != nil {
-		t.Fatalf("status send: %v", err)
 		}
-}
-
-func (p *testPeer) close() {
-	p.pipe.Close()
-}
-
-type fakeTxPool struct {
-	// all transactions are collected.
-	mu  sync.Mutex
-	all []*types.Transaction
-	// if added is non-nil, it receives added transactions.
-	added chan<- []*types.Transaction
-}
-
-func (pool *fakeTxPool) AddTransactions(txs []*types.Transaction) {
-	pool.mu.Lock()
-	defer pool.mu.Unlock()
-	pool.all = append(pool.all, txs...)
-	if pool.added != nil {
-		pool.added <- txs
 	}
-}
-
-func (pool *fakeTxPool) GetTransactions() types.Transactions {
-	pool.mu.Lock()
-	defer pool.mu.Unlock()
-	txs := make([]*types.Transaction, len(pool.all))
-	copy(txs, pool.all)
-	return types.Transactions(txs)
-}
-
-func newtx(from *crypto.Key, nonce uint64, datasize int) *types.Transaction {
-	data := make([]byte, datasize)
-	tx := types.NewTransaction(nonce, common.Address{}, big.NewInt(0), big.NewInt(100000), big.NewInt(0), data)
-	tx, _ = tx.SignECDSA(from.PrivateKey)
-	return tx
 }

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"))