eth, eth/downloader: eth/62 downloader boilerplace + ancestor lookup

eth/downloader/downloader.go

@@ -19,6 +19,7 @@ package downloader
 
 import (
 	"errors"
+	"fmt"
 	"math"
 	"math/big"
 	"sync"
@@ -67,6 +68,7 @@ var (
 	errPendingQueue     = errors.New("pending items in queue")
 	errTimeout          = errors.New("timeout")
 	errEmptyHashSet     = errors.New("empty hash set by peer")
+	errEmptyHeaderSet   = errors.New("empty header set by peer")
 	errPeersUnavailable = errors.New("no peers available or all peers tried for block download process")
 	errAlreadyInPool    = errors.New("hash already in pool")
 	errInvalidChain     = errors.New("retrieved hash chain is invalid")
@@ -91,28 +93,37 @@ type chainInsertFn func(types.Blocks) (int, error)
 // peerDropFn is a callback type for dropping a peer detected as malicious.
 type peerDropFn func(id string)
 
-type blockPack struct {
+// hashPack is a batch of block hashes returned by a peer (eth/61).
-	peerId string
-	blocks []*types.Block
-}
-
 type hashPack struct {
 	peerId string
 	hashes []common.Hash
 }
 
-type crossCheck struct {
+// blockPack is a batch of blocks returned by a peer (eth/61).
-	expire time.Time
+type blockPack struct {
-	parent common.Hash
+	peerId string
+	blocks []*types.Block
+}
+
+// headerPack is a batch of block headers returned by a peer.
+type headerPack struct {
+	peerId  string
+	headers []*types.Header
+}
+
+// bodyPack is a batch of block bodies returned by a peer.
+type bodyPack struct {
+	peerId       string
+	transactions [][]*types.Transaction
+	uncles       [][]*types.Header
 }
 
 type Downloader struct {
 	mux *event.TypeMux
 
-	queue  *queue                      // Scheduler for selecting the hashes to download
+	queue  *queue   // Scheduler for selecting the hashes to download
-	peers  *peerSet                    // Set of active peers from which download can proceed
+	peers  *peerSet // Set of active peers from which download can proceed
-	checks map[common.Hash]*crossCheck // Pending cross checks to verify a hash chain
+	banned *set.Set // Set of hashes we've received and banned
-	banned *set.Set                    // Set of hashes we've received and banned
 
 	interrupt int32 // Atomic boolean to signal termination
 
@@ -137,9 +148,11 @@ type Downloader struct {
 
 	// Channels
 	newPeerCh chan *peer
-	hashCh    chan hashPack  // Channel receiving inbound hashes
+	hashCh    chan hashPack   // [eth/61] Channel receiving inbound hashes
-	blockCh   chan blockPack // Channel receiving inbound blocks
+	blockCh   chan blockPack  // [eth/61] Channel receiving inbound blocks
-	processCh chan bool      // Channel to signal the block fetcher of new or finished work
+	headerCh  chan headerPack // [eth/62] Channel receiving inbound block headers
+	bodyCh    chan bodyPack   // [eth/62] Channel receiving inbound block bodies
+	processCh chan bool       // Channel to signal the block fetcher of new or finished work
 
 	cancelCh   chan struct{} // Channel to cancel mid-flight syncs
 	cancelLock sync.RWMutex  // Lock to protect the cancel channel in delivers
@@ -166,6 +179,8 @@ func New(mux *event.TypeMux, hasBlock hashCheckFn, getBlock blockRetrievalFn, he
 		newPeerCh:   make(chan *peer, 1),
 		hashCh:      make(chan hashPack, 1),
 		blockCh:     make(chan blockPack, 1),
+		headerCh:    make(chan headerPack, 1),
+		bodyCh:      make(chan bodyPack, 1),
 		processCh:   make(chan bool, 1),
 	}
 	// Inject all the known bad hashes
@@ -206,7 +221,9 @@ func (d *Downloader) Synchronising() bool {
 
 // RegisterPeer injects a new download peer into the set of block source to be
 // used for fetching hashes and blocks from.
-func (d *Downloader) RegisterPeer(id string, version int, head common.Hash, getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn) error {
+func (d *Downloader) RegisterPeer(id string, version int, head common.Hash,
+	getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn, // eth/61 callbacks, remove when upgrading
+	getRelHeaders relativeHeaderFetcherFn, getAbsHeaders absoluteHeaderFetcherFn, getBlockBodies blockBodyFetcherFn) error {
 	// If the peer wants to send a banned hash, reject
 	if d.banned.Has(head) {
 		glog.V(logger.Debug).Infoln("Register rejected, head hash banned:", id)
@@ -214,7 +231,7 @@ func (d *Downloader) RegisterPeer(id string, version int, head common.Hash, getR
 	}
 	// Otherwise try to construct and register the peer
 	glog.V(logger.Detail).Infoln("Registering peer", id)
-	if err := d.peers.Register(newPeer(id, version, head, getRelHashes, getAbsHashes, getBlocks)); err != nil {
+	if err := d.peers.Register(newPeer(id, version, head, getRelHashes, getAbsHashes, getBlocks, getRelHeaders, getAbsHeaders, getBlockBodies)); err != nil {
 		glog.V(logger.Error).Infoln("Register failed:", err)
 		return err
 	}
@@ -244,7 +261,7 @@ func (d *Downloader) Synchronise(id string, head common.Hash, td *big.Int) {
 	case errBusy:
 		glog.V(logger.Detail).Infof("Synchronisation already in progress")
 
-	case errTimeout, errBadPeer, errStallingPeer, errBannedHead, errEmptyHashSet, errPeersUnavailable, errInvalidChain, errCrossCheckFailed:
+	case errTimeout, errBadPeer, errStallingPeer, errBannedHead, errEmptyHashSet, errEmptyHeaderSet, errPeersUnavailable, errInvalidChain, errCrossCheckFailed:
 		glog.V(logger.Debug).Infof("Removing peer %v: %v", id, err)
 		d.dropPeer(id)
 
@@ -285,7 +302,6 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int) error
 	// Reset the queue and peer set to clean any internal leftover state
 	d.queue.Reset()
 	d.peers.Reset()
-	d.checks = make(map[common.Hash]*crossCheck)
 
 	// Create cancel channel for aborting mid-flight
 	d.cancelLock.Lock()
@@ -324,13 +340,13 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
 	switch p.version {
 	case eth61:
 		// Old eth/61, use forward, concurrent hash and block retrieval algorithm
-		number, err := d.findAncestor(p)
+		number, err := d.findAncestor61(p)
 		if err != nil {
 			return err
 		}
 		errc := make(chan error, 2)
-		go func() { errc <- d.fetchHashes(p, td, number+1) }()
+		go func() { errc <- d.fetchHashes61(p, td, number+1) }()
-		go func() { errc <- d.fetchBlocks(number + 1) }()
+		go func() { errc <- d.fetchBlocks61(number + 1) }()
 
 		// If any fetcher fails, cancel the other
 		if err := <-errc; err != nil {
@@ -340,6 +356,16 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
 		}
 		return <-errc
 
+	case eth62:
+		// New eth/62, use forward, concurrent header and block body retrieval algorithm
+		fmt.Println("Looking for common ancestor:", p)
+		number, err := d.findAncestor(p)
+		if err != nil {
+			return err
+		}
+		fmt.Println("Common ancestor:", number)
+		return errBadPeer
+
 	default:
 		// Something very wrong, stop right here
 		glog.V(logger.Error).Infof("Unsupported eth protocol: %d", p.version)
@@ -373,12 +399,12 @@ func (d *Downloader) Terminate() {
 	d.cancel()
 }
 
-// findAncestor tries to locate the common ancestor block of the local chain and
+// findAncestor61 tries to locate the common ancestor block of the local chain and
 // a remote peers blockchain. In the general case when our node was in sync and
 // on the correct chain, checking the top N blocks should already get us a match.
 // In the rare scenario when we ended up on a long soft fork (i.e. none of the
 // head blocks match), we do a binary search to find the common ancestor.
-func (d *Downloader) findAncestor(p *peer) (uint64, error) {
+func (d *Downloader) findAncestor61(p *peer) (uint64, error) {
 	glog.V(logger.Debug).Infof("%v: looking for common ancestor", p)
 
 	// Request out head blocks to short circuit ancestor location
@@ -422,6 +448,12 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) {
 		case <-d.blockCh:
 			// Out of bounds blocks received, ignore them
 
+		case <-d.headerCh:
+			// Out of bounds eth/62 block headers received, ignore them
+
+		case <-d.bodyCh:
+			// Out of bounds eth/62 block bodies received, ignore them
+
 		case <-timeout:
 			glog.V(logger.Debug).Infof("%v: head hash timeout", p)
 			return 0, errTimeout
@@ -476,6 +508,12 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) {
 			case <-d.blockCh:
 				// Out of bounds blocks received, ignore them
 
+			case <-d.headerCh:
+				// Out of bounds eth/62 block headers received, ignore them
+
+			case <-d.bodyCh:
+				// Out of bounds eth/62 block bodies received, ignore them
+
 			case <-timeout:
 				glog.V(logger.Debug).Infof("%v: search hash timeout", p)
 				return 0, errTimeout
@@ -485,9 +523,9 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) {
 	return start, nil
 }
 
-// fetchHashes keeps retrieving hashes from the requested number, until no more
+// fetchHashes61 keeps retrieving hashes from the requested number, until no more
 // are returned, potentially throttling on the way.
-func (d *Downloader) fetchHashes(p *peer, td *big.Int, from uint64) error {
+func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error {
 	glog.V(logger.Debug).Infof("%v: downloading hashes from #%d", p, from)
 
 	// Create a timeout timer, and the associated hash fetcher
@@ -573,10 +611,10 @@ func (d *Downloader) fetchHashes(p *peer, td *big.Int, from uint64) error {
 	}
 }
 
-// fetchBlocks iteratively downloads the scheduled hashes, taking any available
+// fetchBlocks61 iteratively downloads the scheduled hashes, taking any available
 // peers, reserving a chunk of blocks for each, waiting for delivery and also
 // periodically checking for timeouts.
-func (d *Downloader) fetchBlocks(from uint64) error {
+func (d *Downloader) fetchBlocks61(from uint64) error {
 	glog.V(logger.Debug).Infof("Downloading blocks from #%d", from)
 	defer glog.V(logger.Debug).Infof("Block download terminated")
 
@@ -718,6 +756,130 @@ func (d *Downloader) fetchBlocks(from uint64) error {
 	}
 }
 
+// findAncestor tries to locate the common ancestor block of the local chain and
+// a remote peers blockchain. In the general case when our node was in sync and
+// on the correct chain, checking the top N blocks should already get us a match.
+// In the rare scenario when we ended up on a long soft fork (i.e. none of the
+// head blocks match), we do a binary search to find the common ancestor.
+func (d *Downloader) findAncestor(p *peer) (uint64, error) {
+	glog.V(logger.Debug).Infof("%v: looking for common ancestor", p)
+
+	// Request our head blocks to short circuit ancestor location
+	head := d.headBlock().NumberU64()
+	from := int64(head) - int64(MaxHashFetch)
+	if from < 0 {
+		from = 0
+	}
+	go p.getAbsHeaders(uint64(from), MaxHeaderFetch, 0, false)
+
+	// Wait for the remote response to the head fetch
+	number, hash := uint64(0), common.Hash{}
+	timeout := time.After(hashTTL)
+
+	for finished := false; !finished; {
+		select {
+		case <-d.cancelCh:
+			return 0, errCancelHashFetch
+
+		case headerPack := <-d.headerCh:
+			// Discard anything not from the origin peer
+			if headerPack.peerId != p.id {
+				glog.V(logger.Debug).Infof("Received headers from incorrect peer(%s)", headerPack.peerId)
+				break
+			}
+			// Make sure the peer actually gave something valid
+			headers := headerPack.headers
+			if len(headers) == 0 {
+				glog.V(logger.Debug).Infof("%v: empty head hash set", p)
+				return 0, errEmptyHashSet
+			}
+			// Check if a common ancestor was found
+			finished = true
+			for i := len(headers) - 1; i >= 0; i-- {
+				if d.hasBlock(headers[i].Hash()) {
+					number, hash = headers[i].Number.Uint64(), headers[i].Hash()
+					break
+				}
+			}
+
+		case <-d.bodyCh:
+			// Out of bounds block bodies received, ignore them
+
+		case <-d.hashCh:
+			// Out of bounds eth/61 hashes received, ignore them
+
+		case <-d.blockCh:
+			// Out of bounds eth/61 blocks received, ignore them
+
+		case <-timeout:
+			glog.V(logger.Debug).Infof("%v: head hash timeout", p)
+			return 0, errTimeout
+		}
+	}
+	// If the head fetch already found an ancestor, return
+	if !common.EmptyHash(hash) {
+		glog.V(logger.Debug).Infof("%v: common ancestor: #%d [%x]", p, number, hash[:4])
+		return number, nil
+	}
+	// Ancestor not found, we need to binary search over our chain
+	start, end := uint64(0), head
+	for start+1 < end {
+		// Split our chain interval in two, and request the hash to cross check
+		check := (start + end) / 2
+
+		timeout := time.After(hashTTL)
+		go p.getAbsHeaders(uint64(check), 1, 0, false)
+
+		// Wait until a reply arrives to this request
+		for arrived := false; !arrived; {
+			select {
+			case <-d.cancelCh:
+				return 0, errCancelHashFetch
+
+			case headerPack := <-d.headerCh:
+				// Discard anything not from the origin peer
+				if headerPack.peerId != p.id {
+					glog.V(logger.Debug).Infof("Received hashes from incorrect peer(%s)", headerPack.peerId)
+					break
+				}
+				// Make sure the peer actually gave something valid
+				headers := headerPack.headers
+				if len(headers) != 1 {
+					glog.V(logger.Debug).Infof("%v: invalid search header set (%d)", p, len(headers))
+					return 0, errBadPeer
+				}
+				arrived = true
+
+				// Modify the search interval based on the response
+				block := d.getBlock(headers[0].Hash())
+				if block == nil {
+					end = check
+					break
+				}
+				if block.NumberU64() != check {
+					glog.V(logger.Debug).Infof("%v: non requested header #%d [%x], instead of #%d", p, block.NumberU64(), block.Hash().Bytes()[:4], check)
+					return 0, errBadPeer
+				}
+				start = check
+
+			case <-d.bodyCh:
+				// Out of bounds block bodies received, ignore them
+
+			case <-d.hashCh:
+				// Out of bounds eth/61 hashes received, ignore them
+
+			case <-d.blockCh:
+				// Out of bounds eth/61 blocks received, ignore them
+
+			case <-timeout:
+				glog.V(logger.Debug).Infof("%v: search hash timeout", p)
+				return 0, errTimeout
+			}
+		}
+	}
+	return start, nil
+}
+
 // process takes blocks from the queue and tries to import them into the chain.
 //
 // The algorithmic flow is as follows:
@@ -796,9 +958,31 @@ func (d *Downloader) process() {
 	}
 }
 
-// DeliverBlocks injects a new batch of blocks received from a remote node.
+// DeliverHashes61 injects a new batch of hashes received from a remote node into
+// the download schedule. This is usually invoked through the BlockHashesMsg by
+// the protocol handler.
+func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) error {
+	// Make sure the downloader is active
+	if atomic.LoadInt32(&d.synchronising) == 0 {
+		return errNoSyncActive
+	}
+	// Deliver or abort if the sync is canceled while queuing
+	d.cancelLock.RLock()
+	cancel := d.cancelCh
+	d.cancelLock.RUnlock()
+
+	select {
+	case d.hashCh <- hashPack{id, hashes}:
+		return nil
+
+	case <-cancel:
+		return errNoSyncActive
+	}
+}
+
+// DeliverBlocks61 injects a new batch of blocks received from a remote node.
 // This is usually invoked through the BlocksMsg by the protocol handler.
-func (d *Downloader) DeliverBlocks(id string, blocks []*types.Block) error {
+func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) error {
 	// Make sure the downloader is active
 	if atomic.LoadInt32(&d.synchronising) == 0 {
 		return errNoSyncActive
@@ -817,10 +1001,9 @@ func (d *Downloader) DeliverBlocks(id string, blocks []*types.Block) error {
 	}
 }
 
-// DeliverHashes injects a new batch of hashes received from a remote node into
+// DeliverHeaders injects a new batch of blck headers received from a remote
-// the download schedule. This is usually invoked through the BlockHashesMsg by
+// node into the download schedule.
-// the protocol handler.
+func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error {
-func (d *Downloader) DeliverHashes(id string, hashes []common.Hash) error {
 	// Make sure the downloader is active
 	if atomic.LoadInt32(&d.synchronising) == 0 {
 		return errNoSyncActive
@@ -831,7 +1014,27 @@ func (d *Downloader) DeliverHashes(id string, hashes []common.Hash) error {
 	d.cancelLock.RUnlock()
 
 	select {
-	case d.hashCh <- hashPack{id, hashes}:
+	case d.headerCh <- headerPack{id, headers}:
+		return nil
+
+	case <-cancel:
+		return errNoSyncActive
+	}
+}
+
+// DeliverBodies injects a new batch of block bodies received from a remote node.
+func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) error {
+	// Make sure the downloader is active
+	if atomic.LoadInt32(&d.synchronising) == 0 {
+		return errNoSyncActive
+	}
+	// Deliver or abort if the sync is canceled while queuing
+	d.cancelLock.RLock()
+	cancel := d.cancelCh
+	d.cancelLock.RUnlock()
+
+	select {
+	case d.bodyCh <- bodyPack{id, transactions, uncles}:
 		return nil
 
 	case <-cancel:

eth/downloader/downloader_test.go

@@ -78,8 +78,6 @@ type downloadTester struct {
 	ownBlocks  map[common.Hash]*types.Block            // Blocks belonging to the tester
 	peerHashes map[string][]common.Hash                // Hash chain belonging to different test peers
 	peerBlocks map[string]map[common.Hash]*types.Block // Blocks belonging to different test peers
-
-	maxHashFetch int // Overrides the maximum number of retrieved hashes
 }
 
 // newTester creates a new downloader test mocker.
@@ -156,7 +154,9 @@ func (dl *downloadTester) newPeer(id string, version int, hashes []common.Hash,
 // specific delay time on processing the network packets sent to it, simulating
 // potentially slow network IO.
 func (dl *downloadTester) newSlowPeer(id string, version int, hashes []common.Hash, blocks map[common.Hash]*types.Block, delay time.Duration) error {
-	err := dl.downloader.RegisterPeer(id, version, hashes[0], dl.peerGetRelHashesFn(id, delay), dl.peerGetAbsHashesFn(id, version, delay), dl.peerGetBlocksFn(id, delay))
+	err := dl.downloader.RegisterPeer(id, version, hashes[0],
+		dl.peerGetRelHashesFn(id, delay), dl.peerGetAbsHashesFn(id, version, delay), dl.peerGetBlocksFn(id, delay),
+		nil, dl.peerGetAbsHeadersFn(id, version, delay), nil)
 	if err == nil {
 		// Assign the owned hashes and blocks to the peer (deep copy)
 		dl.peerHashes[id] = make([]common.Hash, len(hashes))
@@ -184,13 +184,9 @@ func (dl *downloadTester) peerGetRelHashesFn(id string, delay time.Duration) fun
 	return func(head common.Hash) error {
 		time.Sleep(delay)
 
-		limit := MaxHashFetch
-		if dl.maxHashFetch > 0 {
-			limit = dl.maxHashFetch
-		}
 		// Gather the next batch of hashes
 		hashes := dl.peerHashes[id]
-		result := make([]common.Hash, 0, limit)
+		result := make([]common.Hash, 0, MaxHashFetch)
 		for i, hash := range hashes {
 			if hash == head {
 				i++
@@ -204,7 +200,7 @@ func (dl *downloadTester) peerGetRelHashesFn(id string, delay time.Duration) fun
 		// Delay delivery a bit to allow attacks to unfold
 		go func() {
 			time.Sleep(time.Millisecond)
-			dl.downloader.DeliverHashes(id, result)
+			dl.downloader.DeliverHashes61(id, result)
 		}()
 		return nil
 	}
@@ -217,20 +213,16 @@ func (dl *downloadTester) peerGetAbsHashesFn(id string, version int, delay time.
 	return func(head uint64, count int) error {
 		time.Sleep(delay)
 
-		limit := count
-		if dl.maxHashFetch > 0 {
-			limit = dl.maxHashFetch
-		}
 		// Gather the next batch of hashes
 		hashes := dl.peerHashes[id]
-		result := make([]common.Hash, 0, limit)
+		result := make([]common.Hash, 0, count)
-		for i := 0; i < limit && len(hashes)-int(head)-1-i >= 0; i++ {
+		for i := 0; i < count && len(hashes)-int(head)-1-i >= 0; i++ {
 			result = append(result, hashes[len(hashes)-int(head)-1-i])
 		}
 		// Delay delivery a bit to allow attacks to unfold
 		go func() {
 			time.Sleep(time.Millisecond)
-			dl.downloader.DeliverHashes(id, result)
+			dl.downloader.DeliverHashes61(id, result)
 		}()
 		return nil
 	}
@@ -249,22 +241,48 @@ func (dl *downloadTester) peerGetBlocksFn(id string, delay time.Duration) func([
 				result = append(result, block)
 			}
 		}
-		go dl.downloader.DeliverBlocks(id, result)
+		go dl.downloader.DeliverBlocks61(id, result)
 
 		return nil
 	}
 }
 
+// peerGetAbsHeadersFn constructs a GetBlockHeaders function based on a numbered
+// origin; associated with a particular peer in the download tester. The returned
+// function can be used to retrieve batches of headers from the particular peer.
+func (dl *downloadTester) peerGetAbsHeadersFn(id string, version int, delay time.Duration) func(uint64, int, int, bool) error {
+	return func(origin uint64, amount int, skip int, reverse bool) error {
+		time.Sleep(delay)
+
+		// Gather the next batch of hashes
+		hashes := dl.peerHashes[id]
+		blocks := dl.peerBlocks[id]
+		result := make([]*types.Header, 0, amount)
+		for i := 0; i < amount && len(hashes)-int(origin)-1-i >= 0; i++ {
+			result = append(result, blocks[hashes[len(hashes)-int(origin)-1-i]].Header())
+		}
+		// Delay delivery a bit to allow attacks to unfold
+		go func() {
+			time.Sleep(time.Millisecond)
+			dl.downloader.DeliverHeaders(id, result)
+		}()
+		return nil
+	}
+}
+
 // Tests that simple synchronization against a canonical chain works correctly.
 // In this test common ancestor lookup should be short circuited and not require
 // binary searching.
-func TestCanonicalSynchronisation61(t *testing.T) {
+func TestCanonicalSynchronisation61(t *testing.T) { testCanonicalSynchronisation(t, 61) }
+func TestCanonicalSynchronisation62(t *testing.T) { testCanonicalSynchronisation(t, 62) }
+
+func testCanonicalSynchronisation(t *testing.T, protocol int) {
 	// Create a small enough block chain to download
 	targetBlocks := blockCacheLimit - 15
 	hashes, blocks := makeChain(targetBlocks, 0, genesis)
 
 	tester := newTester()
-	tester.newPeer("peer", eth61, hashes, blocks)
+	tester.newPeer("peer", protocol, hashes, blocks)
 
 	// Synchronise with the peer and make sure all blocks were retrieved
 	if err := tester.sync("peer", nil); err != nil {
@@ -336,14 +354,17 @@ func testThrottling(t *testing.T, protocol int) {
 // Tests that simple synchronization against a forked chain works correctly. In
 // this test common ancestor lookup should *not* be short circuited, and a full
 // binary search should be executed.
-func TestForkedSynchronisation61(t *testing.T) {
+func TestForkedSynchronisation61(t *testing.T) { testForkedSynchronisation(t, 61) }
+func TestForkedSynchronisation62(t *testing.T) { testForkedSynchronisation(t, 62) }
+
+func testForkedSynchronisation(t *testing.T, protocol int) {
 	// Create a long enough forked chain
 	common, fork := MaxHashFetch, 2*MaxHashFetch
 	hashesA, hashesB, blocksA, blocksB := makeChainFork(common+fork, fork, genesis)
 
 	tester := newTester()
-	tester.newPeer("fork A", eth61, hashesA, blocksA)
+	tester.newPeer("fork A", protocol, hashesA, blocksA)
-	tester.newPeer("fork B", eth61, hashesB, blocksB)
+	tester.newPeer("fork B", protocol, hashesB, blocksB)
 
 	// Synchronise with the peer and make sure all blocks were retrieved
 	if err := tester.sync("fork A", nil); err != nil {
@@ -366,10 +387,10 @@ func TestInactiveDownloader(t *testing.T) {
 	tester := newTester()
 
 	// Check that neither hashes nor blocks are accepted
-	if err := tester.downloader.DeliverHashes("bad peer", []common.Hash{}); err != errNoSyncActive {
+	if err := tester.downloader.DeliverHashes61("bad peer", []common.Hash{}); err != errNoSyncActive {
 		t.Errorf("error mismatch: have %v, want %v", err, errNoSyncActive)
 	}
-	if err := tester.downloader.DeliverBlocks("bad peer", []*types.Block{}); err != errNoSyncActive {
+	if err := tester.downloader.DeliverBlocks61("bad peer", []*types.Block{}); err != errNoSyncActive {
 		t.Errorf("error mismatch: have %v, want %v", err, errNoSyncActive)
 	}
 }
@@ -463,6 +484,7 @@ func TestHashAttackerDropping(t *testing.T) {
 		{errPendingQueue, false},     // There are blocks still cached, wait to exhaust, no issue
 		{errTimeout, true},           // No hashes received in due time, drop the peer
 		{errEmptyHashSet, true},      // No hashes were returned as a response, drop as it's a dead end
+		{errEmptyHeaderSet, true},    // No headers were returned as a response, drop as it's a dead end
 		{errPeersUnavailable, true},  // Nobody had the advertised blocks, drop the advertiser
 		{errInvalidChain, true},      // Hash chain was detected as invalid, definitely drop
 		{errCrossCheckFailed, true},  // Hash-origin failed to pass a block cross check, drop

eth/downloader/peer.go

@@ -31,10 +31,16 @@ import (
 	"gopkg.in/fatih/set.v0"
 )
 
+// Hash and block fetchers belonging to eth/61 and below
 type relativeHashFetcherFn func(common.Hash) error
 type absoluteHashFetcherFn func(uint64, int) error
 type blockFetcherFn func([]common.Hash) error
 
+// Block header and body fethers belonging to eth/62 and above
+type relativeHeaderFetcherFn func(common.Hash, int, int, bool) error
+type absoluteHeaderFetcherFn func(uint64, int, int, bool) error
+type blockBodyFetcherFn func([]common.Hash) error
+
 var (
 	errAlreadyFetching   = errors.New("already fetching blocks from peer")
 	errAlreadyRegistered = errors.New("peer is already registered")
@@ -54,25 +60,37 @@ type peer struct {
 
 	ignored *set.Set // Set of hashes not to request (didn't have previously)
 
-	getRelHashes relativeHashFetcherFn // Method to retrieve a batch of hashes from an origin hash
+	getRelHashes relativeHashFetcherFn // [eth/61] Method to retrieve a batch of hashes from an origin hash
-	getAbsHashes absoluteHashFetcherFn // Method to retrieve a batch of hashes from an absolute position
+	getAbsHashes absoluteHashFetcherFn // [eth/61] Method to retrieve a batch of hashes from an absolute position
-	getBlocks    blockFetcherFn        // Method to retrieve a batch of blocks
+	getBlocks    blockFetcherFn        // [eth/61] Method to retrieve a batch of blocks
+
+	getRelHeaders  relativeHeaderFetcherFn // [eth/62] Method to retrieve a batch of headers from an origin hash
+	getAbsHeaders  absoluteHeaderFetcherFn // [eth/62] Method to retrieve a batch of headers from an absolute position
+	getBlockBodies blockBodyFetcherFn      // [eth/62] Method to retrieve a batch of block bodies
 
 	version int // Eth protocol version number to switch strategies
 }
 
 // newPeer create a new downloader peer, with specific hash and block retrieval
 // mechanisms.
-func newPeer(id string, version int, head common.Hash, getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn) *peer {
+func newPeer(id string, version int, head common.Hash,
+	getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn, // eth/61 callbacks, remove when upgrading
+	getRelHeaders relativeHeaderFetcherFn, getAbsHeaders absoluteHeaderFetcherFn, getBlockBodies blockBodyFetcherFn) *peer {
 	return &peer{
-		id:           id,
+		id:       id,
-		head:         head,
+		head:     head,
-		capacity:     1,
+		capacity: 1,
+		ignored:  set.New(),
+
 		getRelHashes: getRelHashes,
 		getAbsHashes: getAbsHashes,
 		getBlocks:    getBlocks,
-		ignored:      set.New(),
+
-		version:      version,
+		getRelHeaders:  getRelHeaders,
+		getAbsHeaders:  getAbsHeaders,
+		getBlockBodies: getBlockBodies,
+
+		version: version,
 	}
 }
 

eth/handler.go

@@ -201,7 +201,9 @@ func (pm *ProtocolManager) handle(p *peer) error {
 	defer pm.removePeer(p.id)
 
 	// Register the peer in the downloader. If the downloader considers it banned, we disconnect
-	if err := pm.downloader.RegisterPeer(p.id, p.version, p.Head(), p.RequestHashes, p.RequestHashesFromNumber, p.RequestBlocks); err != nil {
+	if err := pm.downloader.RegisterPeer(p.id, p.version, p.Head(),
+		p.RequestHashes, p.RequestHashesFromNumber, p.RequestBlocks,
+		p.RequestHeadersByHash, p.RequestHeadersByNumber, nil); err != nil {
 		return err
 	}
 	// Propagate existing transactions. new transactions appearing
@@ -287,7 +289,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			break
 		}
 		// Deliver them all to the downloader for queuing
-		err := pm.downloader.DeliverHashes(p.id, hashes)
+		err := pm.downloader.DeliverHashes61(p.id, hashes)
 		if err != nil {
 			glog.V(logger.Debug).Infoln(err)
 		}
@@ -333,7 +335,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		// Filter out any explicitly requested blocks, deliver the rest to the downloader
 		if blocks := pm.fetcher.Filter(blocks); len(blocks) > 0 {
-			pm.downloader.DeliverBlocks(p.id, blocks)
+			pm.downloader.DeliverBlocks61(p.id, blocks)
 		}
 
 	// Block header query, collect the requested headers and reply
@@ -399,8 +401,21 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				query.Origin.Number += (query.Skip + 1)
 			}
 		}
+		fmt.Println("Sending headers", len(headers))
 		return p.SendBlockHeaders(headers)
 
+	case p.version >= eth62 && msg.Code == BlockHeadersMsg:
+		// A batch of headers arrived to one of our previous requests
+		var headers []*types.Header
+		if err := msg.Decode(&headers); err != nil {
+			return errResp(ErrDecode, "msg %v: %v", msg, err)
+		}
+		// Deliver them all to the downloader for queuing
+		err := pm.downloader.DeliverHeaders(p.id, headers)
+		if err != nil {
+			glog.V(logger.Debug).Infoln(err)
+		}
+
 	case p.version >= eth62 && msg.Code == GetBlockBodiesMsg:
 		// Decode the retrieval message
 		msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))

eth/peer.go

@@ -185,40 +185,47 @@ func (p *peer) SendReceipts(receipts []*types.Receipt) error {
 // 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 {
-	glog.V(logger.Debug).Infof("%v fetching hashes (%d) from %x...\n", p, downloader.MaxHashFetch, from[:4])
+	glog.V(logger.Debug).Infof("%v fetching hashes (%d) from %x...", p, downloader.MaxHashFetch, from[:4])
 	return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesData{from, uint64(downloader.MaxHashFetch)})
 }
 
 // RequestHashesFromNumber fetches a batch of hashes from a peer, starting at
 // the requested block number, going upwards towards the genesis block.
 func (p *peer) RequestHashesFromNumber(from uint64, count int) error {
-	glog.V(logger.Debug).Infof("%v fetching hashes (%d) from #%d...\n", p, count, from)
+	glog.V(logger.Debug).Infof("%v fetching hashes (%d) from #%d...", p, count, from)
 	return p2p.Send(p.rw, GetBlockHashesFromNumberMsg, getBlockHashesFromNumberData{from, uint64(count)})
 }
 
 // RequestBlocks fetches a batch of blocks corresponding to the specified hashes.
 func (p *peer) RequestBlocks(hashes []common.Hash) error {
-	glog.V(logger.Debug).Infof("%v fetching %v blocks\n", p, len(hashes))
+	glog.V(logger.Debug).Infof("%v fetching %v blocks", p, len(hashes))
 	return p2p.Send(p.rw, GetBlocksMsg, hashes)
 }
 
-// RequestHeaders fetches a batch of blocks' headers corresponding to the
+// RequestHeadersByHash fetches a batch of blocks' headers corresponding to the
-// specified hashes.
+// specified header query, based on the hash of an origin block.
-func (p *peer) RequestHeaders(hashes []common.Hash) error {
+func (p *peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error {
-	glog.V(logger.Debug).Infof("%v fetching %v headers\n", p, len(hashes))
+	glog.V(logger.Debug).Infof("%v fetching %d headers from %x, skipping %d (reverse = )", p, amount, origin[:4], skip, reverse)
-	return p2p.Send(p.rw, GetBlockHeadersMsg, hashes)
+	return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
+}
+
+// RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the
+// specified header query, based on the number of an origin block.
+func (p *peer) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error {
+	glog.V(logger.Debug).Infof("%v fetching %d headers from #%d, skipping %d (reverse = )", p, amount, origin, skip, reverse)
+	return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
 }
 
 // RequestNodeData fetches a batch of arbitrary data from a node's known state
 // data, corresponding to the specified hashes.
 func (p *peer) RequestNodeData(hashes []common.Hash) error {
-	glog.V(logger.Debug).Infof("%v fetching %v state data\n", p, len(hashes))
+	glog.V(logger.Debug).Infof("%v fetching %v state data", p, len(hashes))
 	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))
+	glog.V(logger.Debug).Infof("%v fetching %v receipts", p, len(hashes))
 	return p2p.Send(p.rw, GetReceiptsMsg, hashes)
 }