From 089257a3da86184c66f666ff8e119e1ca7850357 Mon Sep 17 00:00:00 2001 From: Sonic Date: Fri, 9 Nov 2018 14:47:57 +0800 Subject: dex: copy fetcher and downloader from eth --- dex/fetcher/fetcher.go | 736 +++++++++++++++++++++++++++++++++++++++++ dex/fetcher/fetcher_test.go | 790 ++++++++++++++++++++++++++++++++++++++++++++ dex/fetcher/metrics.go | 43 +++ 3 files changed, 1569 insertions(+) create mode 100644 dex/fetcher/fetcher.go create mode 100644 dex/fetcher/fetcher_test.go create mode 100644 dex/fetcher/metrics.go (limited to 'dex/fetcher') diff --git a/dex/fetcher/fetcher.go b/dex/fetcher/fetcher.go new file mode 100644 index 000000000..f6807b5a5 --- /dev/null +++ b/dex/fetcher/fetcher.go @@ -0,0 +1,736 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package fetcher contains the block announcement based synchronisation. +package fetcher + +import ( + "errors" + "math/rand" + "time" + + "github.com/dexon-foundation/dexon/common" + "github.com/dexon-foundation/dexon/common/prque" + "github.com/dexon-foundation/dexon/consensus" + "github.com/dexon-foundation/dexon/core/types" + "github.com/dexon-foundation/dexon/log" +) + +const ( + arriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested + gatherSlack = 100 * time.Millisecond // Interval used to collate almost-expired announces with fetches + fetchTimeout = 5 * time.Second // Maximum allotted time to return an explicitly requested block + maxUncleDist = 7 // Maximum allowed backward distance from the chain head + maxQueueDist = 32 // Maximum allowed distance from the chain head to queue + hashLimit = 256 // Maximum number of unique blocks a peer may have announced + blockLimit = 64 // Maximum number of unique blocks a peer may have delivered +) + +var ( + errTerminated = errors.New("terminated") +) + +// blockRetrievalFn is a callback type for retrieving a block from the local chain. +type blockRetrievalFn func(common.Hash) *types.Block + +// headerRequesterFn is a callback type for sending a header retrieval request. +type headerRequesterFn func(common.Hash) error + +// bodyRequesterFn is a callback type for sending a body retrieval request. +type bodyRequesterFn func([]common.Hash) error + +// headerVerifierFn is a callback type to verify a block's header for fast propagation. +type headerVerifierFn func(header *types.Header) error + +// blockBroadcasterFn is a callback type for broadcasting a block to connected peers. +type blockBroadcasterFn func(block *types.Block, propagate bool) + +// chainHeightFn is a callback type to retrieve the current chain height. +type chainHeightFn func() uint64 + +// chainInsertFn is a callback type to insert a batch of blocks into the local chain. +type chainInsertFn func(types.Blocks) (int, error) + +// peerDropFn is a callback type for dropping a peer detected as malicious. +type peerDropFn func(id string) + +// announce is the hash notification of the availability of a new block in the +// 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) + header *types.Header // Header of the block partially reassembled (new protocol) + time time.Time // Timestamp of the announcement + + origin string // Identifier of the peer originating the notification + + fetchHeader headerRequesterFn // Fetcher function to retrieve the header of an announced block + fetchBodies bodyRequesterFn // Fetcher function to retrieve the body of an announced block +} + +// headerFilterTask represents a batch of headers needing fetcher filtering. +type headerFilterTask struct { + peer string // The source peer of block headers + headers []*types.Header // Collection of headers to filter + time time.Time // Arrival time of the headers +} + +// bodyFilterTask represents a batch of block bodies (transactions and uncles) +// needing fetcher filtering. +type bodyFilterTask struct { + peer string // The source peer of block bodies + transactions [][]*types.Transaction // Collection of transactions per block bodies + uncles [][]*types.Header // Collection of uncles per block bodies + time time.Time // Arrival time of the blocks' contents +} + +// inject represents a schedules import operation. +type inject struct { + origin string + block *types.Block +} + +// Fetcher is responsible for accumulating block announcements from various peers +// and scheduling them for retrieval. +type Fetcher struct { + // Various event channels + notify chan *announce + inject chan *inject + + blockFilter chan chan []*types.Block + headerFilter chan chan *headerFilterTask + bodyFilter chan chan *bodyFilterTask + + done chan common.Hash + quit chan struct{} + + // Announce states + announces map[string]int // Per peer announce counts to prevent memory exhaustion + announced map[common.Hash][]*announce // Announced blocks, scheduled for fetching + fetching map[common.Hash]*announce // Announced blocks, currently fetching + fetched map[common.Hash][]*announce // Blocks with headers fetched, scheduled for body retrieval + completing map[common.Hash]*announce // Blocks with headers, currently body-completing + + // Block cache + queue *prque.Prque // Queue containing the import operations (block number sorted) + queues map[string]int // Per peer block counts to prevent memory exhaustion + queued map[common.Hash]*inject // Set of already queued blocks (to dedupe imports) + + // Callbacks + getBlock blockRetrievalFn // Retrieves a block from the local chain + verifyHeader headerVerifierFn // Checks if a block's headers have a valid proof of work + broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers + chainHeight chainHeightFn // Retrieves the current chain's height + insertChain chainInsertFn // Injects a batch of blocks into the chain + dropPeer peerDropFn // Drops a peer for misbehaving + + // Testing hooks + announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list + queueChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue + fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch + completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62) + importedHook func(*types.Block) // Method to call upon successful block import (both eth/61 and eth/62) +} + +// New creates a block fetcher to retrieve blocks based on hash announcements. +func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { + return &Fetcher{ + notify: make(chan *announce), + inject: make(chan *inject), + blockFilter: make(chan chan []*types.Block), + headerFilter: make(chan chan *headerFilterTask), + bodyFilter: make(chan chan *bodyFilterTask), + done: make(chan common.Hash), + quit: make(chan struct{}), + announces: make(map[string]int), + announced: make(map[common.Hash][]*announce), + fetching: make(map[common.Hash]*announce), + fetched: make(map[common.Hash][]*announce), + completing: make(map[common.Hash]*announce), + queue: prque.New(nil), + queues: make(map[string]int), + queued: make(map[common.Hash]*inject), + getBlock: getBlock, + verifyHeader: verifyHeader, + broadcastBlock: broadcastBlock, + chainHeight: chainHeight, + insertChain: insertChain, + dropPeer: dropPeer, + } +} + +// Start boots up the announcement based synchroniser, accepting and processing +// hash notifications and block fetches until termination requested. +func (f *Fetcher) Start() { + go f.loop() +} + +// Stop terminates the announcement based synchroniser, canceling all pending +// operations. +func (f *Fetcher) Stop() { + close(f.quit) +} + +// Notify announces the fetcher of the potential availability of a new block in +// the network. +func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time, + headerFetcher headerRequesterFn, bodyFetcher bodyRequesterFn) error { + block := &announce{ + hash: hash, + number: number, + time: time, + origin: peer, + fetchHeader: headerFetcher, + fetchBodies: bodyFetcher, + } + select { + case f.notify <- block: + return nil + case <-f.quit: + return errTerminated + } +} + +// Enqueue tries to fill gaps the fetcher's future import queue. +func (f *Fetcher) Enqueue(peer string, block *types.Block) error { + op := &inject{ + origin: peer, + block: block, + } + select { + case f.inject <- op: + return nil + case <-f.quit: + return errTerminated + } +} + +// FilterHeaders extracts all the headers that were explicitly requested by the fetcher, +// returning those that should be handled differently. +func (f *Fetcher) FilterHeaders(peer string, headers []*types.Header, time time.Time) []*types.Header { + log.Trace("Filtering headers", "peer", peer, "headers", len(headers)) + + // Send the filter channel to the fetcher + filter := make(chan *headerFilterTask) + + select { + case f.headerFilter <- filter: + case <-f.quit: + return nil + } + // Request the filtering of the header list + select { + case filter <- &headerFilterTask{peer: peer, headers: headers, time: time}: + case <-f.quit: + return nil + } + // Retrieve the headers remaining after filtering + select { + case task := <-filter: + return task.headers + case <-f.quit: + return nil + } +} + +// FilterBodies extracts all the block bodies that were explicitly requested by +// the fetcher, returning those that should be handled differently. +func (f *Fetcher) FilterBodies(peer string, transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time) ([][]*types.Transaction, [][]*types.Header) { + log.Trace("Filtering bodies", "peer", peer, "txs", len(transactions), "uncles", len(uncles)) + + // Send the filter channel to the fetcher + filter := make(chan *bodyFilterTask) + + select { + case f.bodyFilter <- filter: + case <-f.quit: + return nil, nil + } + // Request the filtering of the body list + select { + case filter <- &bodyFilterTask{peer: peer, transactions: transactions, uncles: uncles, time: time}: + case <-f.quit: + return nil, nil + } + // Retrieve the bodies remaining after filtering + select { + case task := <-filter: + return task.transactions, task.uncles + case <-f.quit: + return nil, nil + } +} + +// Loop is the main fetcher loop, checking and processing various notification +// events. +func (f *Fetcher) loop() { + // Iterate the block fetching until a quit is requested + fetchTimer := time.NewTimer(0) + completeTimer := time.NewTimer(0) + + for { + // Clean up any expired block fetches + for hash, announce := range f.fetching { + if time.Since(announce.time) > fetchTimeout { + f.forgetHash(hash) + } + } + // Import any queued blocks that could potentially fit + height := f.chainHeight() + for !f.queue.Empty() { + op := f.queue.PopItem().(*inject) + hash := op.block.Hash() + if f.queueChangeHook != nil { + f.queueChangeHook(hash, false) + } + // If too high up the chain or phase, continue later + number := op.block.NumberU64() + if number > height+1 { + f.queue.Push(op, -int64(number)) + if f.queueChangeHook != nil { + f.queueChangeHook(hash, true) + } + break + } + // Otherwise if fresh and still unknown, try and import + if number+maxUncleDist < height || f.getBlock(hash) != nil { + f.forgetBlock(hash) + continue + } + f.insert(op.origin, op.block) + } + // Wait for an outside event to occur + select { + case <-f.quit: + // Fetcher terminating, abort all operations + return + + case notification := <-f.notify: + // A block was announced, make sure the peer isn't DOSing us + propAnnounceInMeter.Mark(1) + + count := f.announces[notification.origin] + 1 + if count > hashLimit { + log.Debug("Peer exceeded outstanding announces", "peer", notification.origin, "limit", hashLimit) + propAnnounceDOSMeter.Mark(1) + break + } + // If we have a valid block number, check that it's potentially useful + if notification.number > 0 { + if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { + log.Debug("Peer discarded announcement", "peer", notification.origin, "number", notification.number, "hash", notification.hash, "distance", dist) + propAnnounceDropMeter.Mark(1) + break + } + } + // All is well, schedule the announce if block's not yet downloading + if _, ok := f.fetching[notification.hash]; ok { + break + } + if _, ok := f.completing[notification.hash]; ok { + break + } + f.announces[notification.origin] = count + f.announced[notification.hash] = append(f.announced[notification.hash], notification) + if f.announceChangeHook != nil && len(f.announced[notification.hash]) == 1 { + f.announceChangeHook(notification.hash, true) + } + if len(f.announced) == 1 { + f.rescheduleFetch(fetchTimer) + } + + case op := <-f.inject: + // A direct block insertion was requested, try and fill any pending gaps + propBroadcastInMeter.Mark(1) + f.enqueue(op.origin, op.block) + + case hash := <-f.done: + // A pending import finished, remove all traces of the notification + f.forgetHash(hash) + f.forgetBlock(hash) + + case <-fetchTimer.C: + // At least one block's timer ran out, check for needing retrieval + request := make(map[string][]common.Hash) + + for hash, announces := range f.announced { + if time.Since(announces[0].time) > arriveTimeout-gatherSlack { + // Pick a random peer to retrieve from, reset all others + announce := announces[rand.Intn(len(announces))] + f.forgetHash(hash) + + // If the block still didn't arrive, queue for fetching + if f.getBlock(hash) == nil { + request[announce.origin] = append(request[announce.origin], hash) + f.fetching[hash] = announce + } + } + } + // Send out all block header requests + for peer, hashes := range request { + log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes) + + // Create a closure of the fetch and schedule in on a new thread + fetchHeader, hashes := f.fetching[hashes[0]].fetchHeader, hashes + go func() { + if f.fetchingHook != nil { + f.fetchingHook(hashes) + } + for _, hash := range hashes { + headerFetchMeter.Mark(1) + fetchHeader(hash) // Suboptimal, but protocol doesn't allow batch header retrievals + } + }() + } + // Schedule the next fetch if blocks are still pending + f.rescheduleFetch(fetchTimer) + + case <-completeTimer.C: + // At least one header's timer ran out, retrieve everything + request := make(map[string][]common.Hash) + + for hash, announces := range f.fetched { + // Pick a random peer to retrieve from, reset all others + announce := announces[rand.Intn(len(announces))] + f.forgetHash(hash) + + // If the block still didn't arrive, queue for completion + if f.getBlock(hash) == nil { + request[announce.origin] = append(request[announce.origin], hash) + f.completing[hash] = announce + } + } + // Send out all block body requests + for peer, hashes := range request { + log.Trace("Fetching scheduled bodies", "peer", peer, "list", hashes) + + // Create a closure of the fetch and schedule in on a new thread + if f.completingHook != nil { + f.completingHook(hashes) + } + bodyFetchMeter.Mark(int64(len(hashes))) + go f.completing[hashes[0]].fetchBodies(hashes) + } + // Schedule the next fetch if blocks are still pending + f.rescheduleComplete(completeTimer) + + case filter := <-f.headerFilter: + // Headers arrived from a remote peer. Extract those that were explicitly + // requested by the fetcher, and return everything else so it's delivered + // to other parts of the system. + var task *headerFilterTask + select { + case task = <-filter: + case <-f.quit: + return + } + headerFilterInMeter.Mark(int64(len(task.headers))) + + // Split the batch of headers into unknown ones (to return to the caller), + // known incomplete ones (requiring body retrievals) and completed blocks. + unknown, incomplete, complete := []*types.Header{}, []*announce{}, []*types.Block{} + for _, header := range task.headers { + hash := header.Hash() + + // Filter fetcher-requested headers from other synchronisation algorithms + if announce := f.fetching[hash]; announce != nil && announce.origin == task.peer && f.fetched[hash] == nil && f.completing[hash] == nil && f.queued[hash] == nil { + // If the delivered header does not match the promised number, drop the announcer + if header.Number.Uint64() != announce.number { + log.Trace("Invalid block number fetched", "peer", announce.origin, "hash", header.Hash(), "announced", announce.number, "provided", header.Number) + f.dropPeer(announce.origin) + f.forgetHash(hash) + continue + } + // Only keep if not imported by other means + if f.getBlock(hash) == nil { + announce.header = header + announce.time = task.time + + // If the block is empty (header only), short circuit into the final import queue + if header.TxHash == types.DeriveSha(types.Transactions{}) && header.UncleHash == types.CalcUncleHash([]*types.Header{}) { + log.Trace("Block empty, skipping body retrieval", "peer", announce.origin, "number", header.Number, "hash", header.Hash()) + + block := types.NewBlockWithHeader(header) + block.ReceivedAt = task.time + + complete = append(complete, block) + f.completing[hash] = announce + continue + } + // Otherwise add to the list of blocks needing completion + incomplete = append(incomplete, announce) + } else { + log.Trace("Block already imported, discarding header", "peer", announce.origin, "number", header.Number, "hash", header.Hash()) + f.forgetHash(hash) + } + } else { + // Fetcher doesn't know about it, add to the return list + unknown = append(unknown, header) + } + } + headerFilterOutMeter.Mark(int64(len(unknown))) + select { + case filter <- &headerFilterTask{headers: unknown, time: task.time}: + case <-f.quit: + return + } + // Schedule the retrieved headers for body completion + for _, announce := range incomplete { + hash := announce.header.Hash() + if _, ok := f.completing[hash]; ok { + continue + } + f.fetched[hash] = append(f.fetched[hash], announce) + if len(f.fetched) == 1 { + f.rescheduleComplete(completeTimer) + } + } + // Schedule the header-only blocks for import + for _, block := range complete { + if announce := f.completing[block.Hash()]; announce != nil { + f.enqueue(announce.origin, block) + } + } + + case filter := <-f.bodyFilter: + // Block bodies arrived, extract any explicitly requested blocks, return the rest + var task *bodyFilterTask + select { + case task = <-filter: + case <-f.quit: + return + } + bodyFilterInMeter.Mark(int64(len(task.transactions))) + + blocks := []*types.Block{} + for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ { + // Match up a body to any possible completion request + matched := false + + for hash, announce := range f.completing { + if f.queued[hash] == nil { + txnHash := types.DeriveSha(types.Transactions(task.transactions[i])) + uncleHash := types.CalcUncleHash(task.uncles[i]) + + if txnHash == announce.header.TxHash && uncleHash == announce.header.UncleHash && announce.origin == task.peer { + // Mark the body matched, reassemble if still unknown + matched = true + + if f.getBlock(hash) == nil { + block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i]) + block.ReceivedAt = task.time + + blocks = append(blocks, block) + } else { + f.forgetHash(hash) + } + } + } + } + if matched { + task.transactions = append(task.transactions[:i], task.transactions[i+1:]...) + task.uncles = append(task.uncles[:i], task.uncles[i+1:]...) + i-- + continue + } + } + + bodyFilterOutMeter.Mark(int64(len(task.transactions))) + select { + case filter <- task: + case <-f.quit: + return + } + // Schedule the retrieved blocks for ordered import + for _, block := range blocks { + if announce := f.completing[block.Hash()]; announce != nil { + f.enqueue(announce.origin, block) + } + } + } + } +} + +// rescheduleFetch resets the specified fetch timer to the next announce timeout. +func (f *Fetcher) rescheduleFetch(fetch *time.Timer) { + // Short circuit if no blocks are announced + if len(f.announced) == 0 { + return + } + // Otherwise find the earliest expiring announcement + earliest := time.Now() + for _, announces := range f.announced { + if earliest.After(announces[0].time) { + earliest = announces[0].time + } + } + fetch.Reset(arriveTimeout - time.Since(earliest)) +} + +// rescheduleComplete resets the specified completion timer to the next fetch timeout. +func (f *Fetcher) rescheduleComplete(complete *time.Timer) { + // Short circuit if no headers are fetched + if len(f.fetched) == 0 { + return + } + // Otherwise find the earliest expiring announcement + earliest := time.Now() + for _, announces := range f.fetched { + if earliest.After(announces[0].time) { + earliest = announces[0].time + } + } + complete.Reset(gatherSlack - time.Since(earliest)) +} + +// enqueue schedules a new future import operation, if the block to be imported +// has not yet been seen. +func (f *Fetcher) enqueue(peer string, block *types.Block) { + hash := block.Hash() + + // Ensure the peer isn't DOSing us + count := f.queues[peer] + 1 + if count > blockLimit { + log.Debug("Discarded propagated block, exceeded allowance", "peer", peer, "number", block.Number(), "hash", hash, "limit", blockLimit) + propBroadcastDOSMeter.Mark(1) + f.forgetHash(hash) + return + } + // Discard any past or too distant blocks + if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { + log.Debug("Discarded propagated block, too far away", "peer", peer, "number", block.Number(), "hash", hash, "distance", dist) + propBroadcastDropMeter.Mark(1) + f.forgetHash(hash) + return + } + // Schedule the block for future importing + if _, ok := f.queued[hash]; !ok { + op := &inject{ + origin: peer, + block: block, + } + f.queues[peer] = count + f.queued[hash] = op + f.queue.Push(op, -int64(block.NumberU64())) + if f.queueChangeHook != nil { + f.queueChangeHook(op.block.Hash(), true) + } + log.Debug("Queued propagated block", "peer", peer, "number", block.Number(), "hash", hash, "queued", f.queue.Size()) + } +} + +// insert spawns a new goroutine to run a block insertion into the chain. If the +// block's number is at the same height as the current import phase, it updates +// the phase states accordingly. +func (f *Fetcher) insert(peer string, block *types.Block) { + hash := block.Hash() + + // Run the import on a new thread + log.Debug("Importing propagated block", "peer", peer, "number", block.Number(), "hash", hash) + go func() { + defer func() { f.done <- hash }() + + // If the parent's unknown, abort insertion + parent := f.getBlock(block.ParentHash()) + if parent == nil { + log.Debug("Unknown parent of propagated block", "peer", peer, "number", block.Number(), "hash", hash, "parent", block.ParentHash()) + return + } + // Quickly validate the header and propagate the block if it passes + switch err := f.verifyHeader(block.Header()); err { + case nil: + // All ok, quickly propagate to our peers + propBroadcastOutTimer.UpdateSince(block.ReceivedAt) + go f.broadcastBlock(block, true) + + case consensus.ErrFutureBlock: + // Weird future block, don't fail, but neither propagate + + default: + // Something went very wrong, drop the peer + log.Debug("Propagated block verification failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err) + f.dropPeer(peer) + return + } + // Run the actual import and log any issues + if _, err := f.insertChain(types.Blocks{block}); err != nil { + log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err) + return + } + // If import succeeded, broadcast the block + propAnnounceOutTimer.UpdateSince(block.ReceivedAt) + go f.broadcastBlock(block, false) + + // Invoke the testing hook if needed + if f.importedHook != nil { + f.importedHook(block) + } + }() +} + +// forgetHash removes all traces of a block announcement from the fetcher's +// internal state. +func (f *Fetcher) forgetHash(hash common.Hash) { + // Remove all pending announces and decrement DOS counters + for _, announce := range f.announced[hash] { + f.announces[announce.origin]-- + if f.announces[announce.origin] == 0 { + delete(f.announces, announce.origin) + } + } + delete(f.announced, hash) + if f.announceChangeHook != nil { + f.announceChangeHook(hash, false) + } + // Remove any pending fetches and decrement the DOS counters + if announce := f.fetching[hash]; announce != nil { + f.announces[announce.origin]-- + if f.announces[announce.origin] == 0 { + delete(f.announces, announce.origin) + } + delete(f.fetching, hash) + } + + // Remove any pending completion requests and decrement the DOS counters + for _, announce := range f.fetched[hash] { + f.announces[announce.origin]-- + if f.announces[announce.origin] == 0 { + delete(f.announces, announce.origin) + } + } + delete(f.fetched, hash) + + // Remove any pending completions and decrement the DOS counters + if announce := f.completing[hash]; announce != nil { + f.announces[announce.origin]-- + if f.announces[announce.origin] == 0 { + delete(f.announces, announce.origin) + } + delete(f.completing, hash) + } +} + +// forgetBlock removes all traces of a queued block from the fetcher's internal +// state. +func (f *Fetcher) forgetBlock(hash common.Hash) { + if insert := f.queued[hash]; insert != nil { + f.queues[insert.origin]-- + if f.queues[insert.origin] == 0 { + delete(f.queues, insert.origin) + } + delete(f.queued, hash) + } +} diff --git a/dex/fetcher/fetcher_test.go b/dex/fetcher/fetcher_test.go new file mode 100644 index 000000000..24611a8a0 --- /dev/null +++ b/dex/fetcher/fetcher_test.go @@ -0,0 +1,790 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package fetcher + +import ( + "errors" + "math/big" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/dexon-foundation/dexon/common" + "github.com/dexon-foundation/dexon/consensus/ethash" + "github.com/dexon-foundation/dexon/core" + "github.com/dexon-foundation/dexon/core/types" + "github.com/dexon-foundation/dexon/crypto" + "github.com/dexon-foundation/dexon/ethdb" + "github.com/dexon-foundation/dexon/params" +) + +var ( + testdb = ethdb.NewMemDatabase() + testKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + testAddress = crypto.PubkeyToAddress(testKey.PublicKey) + genesis = core.GenesisBlockForTesting(testdb, testAddress, big.NewInt(1000000000)) + unknownBlock = types.NewBlock(&types.Header{GasLimit: params.GenesisGasLimit}, nil, nil, nil) +) + +// makeChain creates a chain of n blocks starting at and including parent. +// the returned hash chain is ordered head->parent. In addition, every 3rd block +// contains a transaction and every 5th an uncle to allow testing correct block +// reassembly. +func makeChain(n int, seed byte, parent *types.Block) ([]common.Hash, map[common.Hash]*types.Block) { + blocks, _ := core.GenerateChain(params.TestChainConfig, parent, ethash.NewFaker(), testdb, n, func(i int, block *core.BlockGen) { + block.SetCoinbase(common.Address{seed}) + + // If the block number is multiple of 3, send a bonus transaction to the miner + if parent == genesis && i%3 == 0 { + signer := types.MakeSigner(params.TestChainConfig, block.Number()) + tx, err := types.SignTx(types.NewTransaction(block.TxNonce(testAddress), common.Address{seed}, big.NewInt(1000), params.TxGas, nil, nil), signer, testKey) + if err != nil { + panic(err) + } + block.AddTx(tx) + } + // If the block number is a multiple of 5, add a bonus uncle to the block + if i%5 == 0 { + block.AddUncle(&types.Header{ParentHash: block.PrevBlock(i - 1).Hash(), Number: big.NewInt(int64(i - 1))}) + } + }) + hashes := make([]common.Hash, n+1) + hashes[len(hashes)-1] = parent.Hash() + blockm := make(map[common.Hash]*types.Block, n+1) + blockm[parent.Hash()] = parent + for i, b := range blocks { + hashes[len(hashes)-i-2] = b.Hash() + blockm[b.Hash()] = b + } + return hashes, blockm +} + +// fetcherTester is a test simulator for mocking out local block chain. +type fetcherTester struct { + fetcher *Fetcher + + hashes []common.Hash // Hash chain belonging to the tester + blocks map[common.Hash]*types.Block // Blocks belonging to the tester + drops map[string]bool // Map of peers dropped by the fetcher + + lock sync.RWMutex +} + +// newTester creates a new fetcher test mocker. +func newTester() *fetcherTester { + tester := &fetcherTester{ + hashes: []common.Hash{genesis.Hash()}, + blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis}, + drops: make(map[string]bool), + } + tester.fetcher = New(tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer) + tester.fetcher.Start() + + return tester +} + +// getBlock retrieves a block from the tester's block chain. +func (f *fetcherTester) getBlock(hash common.Hash) *types.Block { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.blocks[hash] +} + +// verifyHeader is a nop placeholder for the block header verification. +func (f *fetcherTester) verifyHeader(header *types.Header) error { + return nil +} + +// broadcastBlock is a nop placeholder for the block broadcasting. +func (f *fetcherTester) broadcastBlock(block *types.Block, propagate bool) { +} + +// chainHeight retrieves the current height (block number) of the chain. +func (f *fetcherTester) chainHeight() uint64 { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() +} + +// insertChain injects a new blocks into the simulated chain. +func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) { + f.lock.Lock() + defer f.lock.Unlock() + + for i, block := range blocks { + // Make sure the parent in known + if _, ok := f.blocks[block.ParentHash()]; !ok { + return i, errors.New("unknown parent") + } + // Discard any new blocks if the same height already exists + if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() { + return i, nil + } + // Otherwise build our current chain + f.hashes = append(f.hashes, block.Hash()) + f.blocks[block.Hash()] = block + } + return 0, nil +} + +// dropPeer is an emulator for the peer removal, simply accumulating the various +// peers dropped by the fetcher. +func (f *fetcherTester) dropPeer(peer string) { + f.lock.Lock() + defer f.lock.Unlock() + + f.drops[peer] = true +} + +// makeHeaderFetcher retrieves a block header fetcher associated with a simulated peer. +func (f *fetcherTester) makeHeaderFetcher(peer string, blocks map[common.Hash]*types.Block, drift time.Duration) headerRequesterFn { + closure := make(map[common.Hash]*types.Block) + for hash, block := range blocks { + closure[hash] = block + } + // Create a function that return a header from the closure + return func(hash common.Hash) error { + // Gather the blocks to return + headers := make([]*types.Header, 0, 1) + if block, ok := closure[hash]; ok { + headers = append(headers, block.Header()) + } + // Return on a new thread + go f.fetcher.FilterHeaders(peer, headers, time.Now().Add(drift)) + + return nil + } +} + +// makeBodyFetcher retrieves a block body fetcher associated with a simulated peer. +func (f *fetcherTester) makeBodyFetcher(peer string, blocks map[common.Hash]*types.Block, drift time.Duration) bodyRequesterFn { + closure := make(map[common.Hash]*types.Block) + for hash, block := range blocks { + closure[hash] = block + } + // Create a function that returns blocks from the closure + return func(hashes []common.Hash) error { + // Gather the block bodies to return + transactions := make([][]*types.Transaction, 0, len(hashes)) + uncles := make([][]*types.Header, 0, len(hashes)) + + for _, hash := range hashes { + if block, ok := closure[hash]; ok { + transactions = append(transactions, block.Transactions()) + uncles = append(uncles, block.Uncles()) + } + } + // Return on a new thread + go f.fetcher.FilterBodies(peer, transactions, uncles, time.Now().Add(drift)) + + return nil + } +} + +// verifyFetchingEvent verifies that one single event arrive on a fetching channel. +func verifyFetchingEvent(t *testing.T, fetching chan []common.Hash, arrive bool) { + if arrive { + select { + case <-fetching: + case <-time.After(time.Second): + t.Fatalf("fetching timeout") + } + } else { + select { + case <-fetching: + t.Fatalf("fetching invoked") + case <-time.After(10 * time.Millisecond): + } + } +} + +// verifyCompletingEvent verifies that one single event arrive on an completing channel. +func verifyCompletingEvent(t *testing.T, completing chan []common.Hash, arrive bool) { + if arrive { + select { + case <-completing: + case <-time.After(time.Second): + t.Fatalf("completing timeout") + } + } else { + select { + case <-completing: + t.Fatalf("completing invoked") + case <-time.After(10 * time.Millisecond): + } + } +} + +// verifyImportEvent verifies that one single event arrive on an import channel. +func verifyImportEvent(t *testing.T, imported chan *types.Block, arrive bool) { + if arrive { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("import timeout") + } + } else { + select { + case <-imported: + t.Fatalf("import invoked") + case <-time.After(10 * time.Millisecond): + } + } +} + +// verifyImportCount verifies that exactly count number of events arrive on an +// import hook channel. +func verifyImportCount(t *testing.T, imported chan *types.Block, count int) { + for i := 0; i < count; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i+1) + } + } + verifyImportDone(t, imported) +} + +// verifyImportDone verifies that no more events are arriving on an import channel. +func verifyImportDone(t *testing.T, imported chan *types.Block) { + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): + } +} + +// Tests that a fetcher accepts block announcements and initiates retrievals for +// them, successfully importing into the local chain. +func TestSequentialAnnouncements62(t *testing.T) { testSequentialAnnouncements(t, 62) } +func TestSequentialAnnouncements63(t *testing.T) { testSequentialAnnouncements(t, 63) } +func TestSequentialAnnouncements64(t *testing.T) { testSequentialAnnouncements(t, 64) } + +func testSequentialAnnouncements(t *testing.T, protocol int) { + // Create a chain of blocks to import + targetBlocks := 4 * hashLimit + hashes, blocks := makeChain(targetBlocks, 0, genesis) + + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + // Iteratively announce blocks until all are imported + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 2; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + verifyImportEvent(t, imported, true) + } + verifyImportDone(t, imported) +} + +// Tests that if blocks are announced by multiple peers (or even the same buggy +// peer), they will only get downloaded at most once. +func TestConcurrentAnnouncements62(t *testing.T) { testConcurrentAnnouncements(t, 62) } +func TestConcurrentAnnouncements63(t *testing.T) { testConcurrentAnnouncements(t, 63) } +func TestConcurrentAnnouncements64(t *testing.T) { testConcurrentAnnouncements(t, 64) } + +func testConcurrentAnnouncements(t *testing.T, protocol int) { + // Create a chain of blocks to import + targetBlocks := 4 * hashLimit + hashes, blocks := makeChain(targetBlocks, 0, genesis) + + // Assemble a tester with a built in counter for the requests + tester := newTester() + firstHeaderFetcher := tester.makeHeaderFetcher("first", blocks, -gatherSlack) + firstBodyFetcher := tester.makeBodyFetcher("first", blocks, 0) + secondHeaderFetcher := tester.makeHeaderFetcher("second", blocks, -gatherSlack) + secondBodyFetcher := tester.makeBodyFetcher("second", blocks, 0) + + counter := uint32(0) + firstHeaderWrapper := func(hash common.Hash) error { + atomic.AddUint32(&counter, 1) + return firstHeaderFetcher(hash) + } + secondHeaderWrapper := func(hash common.Hash) error { + atomic.AddUint32(&counter, 1) + return secondHeaderFetcher(hash) + } + // Iteratively announce blocks until all are imported + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 2; i >= 0; i-- { + tester.fetcher.Notify("first", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), firstHeaderWrapper, firstBodyFetcher) + tester.fetcher.Notify("second", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout+time.Millisecond), secondHeaderWrapper, secondBodyFetcher) + tester.fetcher.Notify("second", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout-time.Millisecond), secondHeaderWrapper, secondBodyFetcher) + verifyImportEvent(t, imported, true) + } + verifyImportDone(t, imported) + + // Make sure no blocks were retrieved twice + if int(counter) != targetBlocks { + t.Fatalf("retrieval count mismatch: have %v, want %v", counter, targetBlocks) + } +} + +// Tests that announcements arriving while a previous is being fetched still +// results in a valid import. +func TestOverlappingAnnouncements62(t *testing.T) { testOverlappingAnnouncements(t, 62) } +func TestOverlappingAnnouncements63(t *testing.T) { testOverlappingAnnouncements(t, 63) } +func TestOverlappingAnnouncements64(t *testing.T) { testOverlappingAnnouncements(t, 64) } + +func testOverlappingAnnouncements(t *testing.T, protocol int) { + // Create a chain of blocks to import + targetBlocks := 4 * hashLimit + hashes, blocks := makeChain(targetBlocks, 0, genesis) + + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + // Iteratively announce blocks, but overlap them continuously + overlap := 16 + imported := make(chan *types.Block, len(hashes)-1) + for i := 0; i < overlap; i++ { + imported <- nil + } + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 2; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", len(hashes)-i) + } + } + // Wait for all the imports to complete and check count + verifyImportCount(t, imported, overlap) +} + +// Tests that announces already being retrieved will not be duplicated. +func TestPendingDeduplication62(t *testing.T) { testPendingDeduplication(t, 62) } +func TestPendingDeduplication63(t *testing.T) { testPendingDeduplication(t, 63) } +func TestPendingDeduplication64(t *testing.T) { testPendingDeduplication(t, 64) } + +func testPendingDeduplication(t *testing.T, protocol int) { + // Create a hash and corresponding block + hashes, blocks := makeChain(1, 0, genesis) + + // Assemble a tester with a built in counter and delayed fetcher + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("repeater", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("repeater", blocks, 0) + + delay := 50 * time.Millisecond + counter := uint32(0) + headerWrapper := func(hash common.Hash) error { + atomic.AddUint32(&counter, 1) + + // Simulate a long running fetch + go func() { + time.Sleep(delay) + headerFetcher(hash) + }() + return nil + } + // 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], 1, time.Now().Add(-arriveTimeout), headerWrapper, bodyFetcher) + time.Sleep(time.Millisecond) + } + time.Sleep(delay) + + // Check that all blocks were imported and none fetched twice + if imported := len(tester.blocks); imported != 2 { + t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 2) + } + if int(counter) != 1 { + t.Fatalf("retrieval count mismatch: have %v, want %v", counter, 1) + } +} + +// Tests that announcements retrieved in a random order are cached and eventually +// imported when all the gaps are filled in. +func TestRandomArrivalImport62(t *testing.T) { testRandomArrivalImport(t, 62) } +func TestRandomArrivalImport63(t *testing.T) { testRandomArrivalImport(t, 63) } +func TestRandomArrivalImport64(t *testing.T) { testRandomArrivalImport(t, 64) } + +func testRandomArrivalImport(t *testing.T, protocol int) { + // Create a chain of blocks to import, and choose one to delay + targetBlocks := maxQueueDist + hashes, blocks := makeChain(targetBlocks, 0, genesis) + skip := targetBlocks / 2 + + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + // Iteratively announce blocks, skipping one entry + imported := make(chan *types.Block, len(hashes)-1) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 1; i >= 0; i-- { + if i != skip { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + time.Sleep(time.Millisecond) + } + } + // Finally announce the skipped entry and check full import + tester.fetcher.Notify("valid", hashes[skip], uint64(len(hashes)-skip-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + verifyImportCount(t, imported, len(hashes)-1) +} + +// Tests that direct block enqueues (due to block propagation vs. hash announce) +// are correctly schedule, filling and import queue gaps. +func TestQueueGapFill62(t *testing.T) { testQueueGapFill(t, 62) } +func TestQueueGapFill63(t *testing.T) { testQueueGapFill(t, 63) } +func TestQueueGapFill64(t *testing.T) { testQueueGapFill(t, 64) } + +func testQueueGapFill(t *testing.T, protocol int) { + // Create a chain of blocks to import, and choose one to not announce at all + targetBlocks := maxQueueDist + hashes, blocks := makeChain(targetBlocks, 0, genesis) + skip := targetBlocks / 2 + + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + // Iteratively announce blocks, skipping one entry + imported := make(chan *types.Block, len(hashes)-1) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 1; i >= 0; i-- { + if i != skip { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + time.Sleep(time.Millisecond) + } + } + // Fill the missing block directly as if propagated + tester.fetcher.Enqueue("valid", blocks[hashes[skip]]) + verifyImportCount(t, imported, len(hashes)-1) +} + +// Tests that blocks arriving from various sources (multiple propagations, hash +// announces, etc) do not get scheduled for import multiple times. +func TestImportDeduplication62(t *testing.T) { testImportDeduplication(t, 62) } +func TestImportDeduplication63(t *testing.T) { testImportDeduplication(t, 63) } +func TestImportDeduplication64(t *testing.T) { testImportDeduplication(t, 64) } + +func testImportDeduplication(t *testing.T, protocol int) { + // Create two blocks to import (one for duplication, the other for stalling) + hashes, blocks := makeChain(2, 0, genesis) + + // Create the tester and wrap the importer with a counter + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + counter := uint32(0) + tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { + atomic.AddUint32(&counter, uint32(len(blocks))) + return tester.insertChain(blocks) + } + // Instrument the fetching and imported events + fetching := make(chan []common.Hash) + imported := make(chan *types.Block, len(hashes)-1) + tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes } + 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], 1, time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + <-fetching + + tester.fetcher.Enqueue("valid", blocks[hashes[0]]) + tester.fetcher.Enqueue("valid", blocks[hashes[0]]) + tester.fetcher.Enqueue("valid", blocks[hashes[0]]) + + // Fill the missing block directly as if propagated, and check import uniqueness + tester.fetcher.Enqueue("valid", blocks[hashes[1]]) + verifyImportCount(t, imported, 2) + + if counter != 2 { + t.Fatalf("import invocation count mismatch: have %v, want %v", counter, 2) + } +} + +// Tests that blocks with numbers much lower or higher than out current head get +// discarded to prevent wasting resources on useless blocks from faulty peers. +func TestDistantPropagationDiscarding(t *testing.T) { + // Create a long chain to import and define the discard boundaries + hashes, blocks := makeChain(3*maxQueueDist, 0, genesis) + head := hashes[len(hashes)/2] + + low, high := len(hashes)/2+maxUncleDist+1, len(hashes)/2-maxQueueDist-1 + + // Create a tester and simulate a head block being the middle of the above chain + tester := newTester() + + tester.lock.Lock() + tester.hashes = []common.Hash{head} + tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} + tester.lock.Unlock() + + // Ensure that a block with a lower number than the threshold is discarded + tester.fetcher.Enqueue("lower", blocks[hashes[low]]) + time.Sleep(10 * time.Millisecond) + if !tester.fetcher.queue.Empty() { + t.Fatalf("fetcher queued stale block") + } + // Ensure that a block with a higher number than the threshold is discarded + tester.fetcher.Enqueue("higher", blocks[hashes[high]]) + time.Sleep(10 * time.Millisecond) + if !tester.fetcher.queue.Empty() { + t.Fatalf("fetcher queued future block") + } +} + +// Tests that announcements with numbers much lower or higher than out current +// head get discarded to prevent wasting resources on useless blocks from faulty +// peers. +func TestDistantAnnouncementDiscarding62(t *testing.T) { testDistantAnnouncementDiscarding(t, 62) } +func TestDistantAnnouncementDiscarding63(t *testing.T) { testDistantAnnouncementDiscarding(t, 63) } +func TestDistantAnnouncementDiscarding64(t *testing.T) { testDistantAnnouncementDiscarding(t, 64) } + +func testDistantAnnouncementDiscarding(t *testing.T, protocol int) { + // Create a long chain to import and define the discard boundaries + hashes, blocks := makeChain(3*maxQueueDist, 0, genesis) + head := hashes[len(hashes)/2] + + low, high := len(hashes)/2+maxUncleDist+1, len(hashes)/2-maxQueueDist-1 + + // Create a tester and simulate a head block being the middle of the above chain + tester := newTester() + + tester.lock.Lock() + tester.hashes = []common.Hash{head} + tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} + tester.lock.Unlock() + + headerFetcher := tester.makeHeaderFetcher("lower", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("lower", blocks, 0) + + fetching := make(chan struct{}, 2) + tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- struct{}{} } + + // Ensure that a block with a lower number than the threshold is discarded + tester.fetcher.Notify("lower", hashes[low], blocks[hashes[low]].NumberU64(), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + select { + case <-time.After(50 * time.Millisecond): + case <-fetching: + t.Fatalf("fetcher requested stale header") + } + // Ensure that a block with a higher number than the threshold is discarded + tester.fetcher.Notify("higher", hashes[high], blocks[hashes[high]].NumberU64(), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + select { + case <-time.After(50 * time.Millisecond): + case <-fetching: + t.Fatalf("fetcher requested future header") + } +} + +// Tests that peers announcing blocks with invalid numbers (i.e. not matching +// the headers provided afterwards) get dropped as malicious. +func TestInvalidNumberAnnouncement62(t *testing.T) { testInvalidNumberAnnouncement(t, 62) } +func TestInvalidNumberAnnouncement63(t *testing.T) { testInvalidNumberAnnouncement(t, 63) } +func TestInvalidNumberAnnouncement64(t *testing.T) { testInvalidNumberAnnouncement(t, 64) } + +func testInvalidNumberAnnouncement(t *testing.T, protocol int) { + // Create a single block to import and check numbers against + hashes, blocks := makeChain(1, 0, genesis) + + tester := newTester() + badHeaderFetcher := tester.makeHeaderFetcher("bad", blocks, -gatherSlack) + badBodyFetcher := tester.makeBodyFetcher("bad", blocks, 0) + + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + // Announce a block with a bad number, check for immediate drop + tester.fetcher.Notify("bad", hashes[0], 2, time.Now().Add(-arriveTimeout), badHeaderFetcher, badBodyFetcher) + verifyImportEvent(t, imported, false) + + tester.lock.RLock() + dropped := tester.drops["bad"] + tester.lock.RUnlock() + + if !dropped { + t.Fatalf("peer with invalid numbered announcement not dropped") + } + + goodHeaderFetcher := tester.makeHeaderFetcher("good", blocks, -gatherSlack) + goodBodyFetcher := tester.makeBodyFetcher("good", blocks, 0) + // Make sure a good announcement passes without a drop + tester.fetcher.Notify("good", hashes[0], 1, time.Now().Add(-arriveTimeout), goodHeaderFetcher, goodBodyFetcher) + verifyImportEvent(t, imported, true) + + tester.lock.RLock() + dropped = tester.drops["good"] + tester.lock.RUnlock() + + if dropped { + t.Fatalf("peer with valid numbered announcement dropped") + } + verifyImportDone(t, imported) +} + +// Tests that if a block is empty (i.e. header only), no body request should be +// made, and instead the header should be assembled into a whole block in itself. +func TestEmptyBlockShortCircuit62(t *testing.T) { testEmptyBlockShortCircuit(t, 62) } +func TestEmptyBlockShortCircuit63(t *testing.T) { testEmptyBlockShortCircuit(t, 63) } +func TestEmptyBlockShortCircuit64(t *testing.T) { testEmptyBlockShortCircuit(t, 64) } + +func testEmptyBlockShortCircuit(t *testing.T, protocol int) { + // Create a chain of blocks to import + hashes, blocks := makeChain(32, 0, genesis) + + tester := newTester() + headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + // Add a monitoring hook for all internal events + fetching := make(chan []common.Hash) + tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes } + + completing := make(chan []common.Hash) + tester.fetcher.completingHook = func(hashes []common.Hash) { completing <- hashes } + + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + // Iteratively announce blocks until all are imported + for i := len(hashes) - 2; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher) + + // All announces should fetch the header + verifyFetchingEvent(t, fetching, true) + + // Only blocks with data contents should request bodies + verifyCompletingEvent(t, completing, len(blocks[hashes[i]].Transactions()) > 0 || len(blocks[hashes[i]].Uncles()) > 0) + + // Irrelevant of the construct, import should succeed + verifyImportEvent(t, imported, true) + } + verifyImportDone(t, imported) +} + +// Tests that a peer is unable to use unbounded memory with sending infinite +// block announcements to a node, but that even in the face of such an attack, +// the fetcher remains operational. +func TestHashMemoryExhaustionAttack62(t *testing.T) { testHashMemoryExhaustionAttack(t, 62) } +func TestHashMemoryExhaustionAttack63(t *testing.T) { testHashMemoryExhaustionAttack(t, 63) } +func TestHashMemoryExhaustionAttack64(t *testing.T) { testHashMemoryExhaustionAttack(t, 64) } + +func testHashMemoryExhaustionAttack(t *testing.T, protocol int) { + // Create a tester with instrumented import hooks + tester := newTester() + + imported, announces := make(chan *types.Block), int32(0) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + tester.fetcher.announceChangeHook = func(hash common.Hash, added bool) { + if added { + atomic.AddInt32(&announces, 1) + } else { + atomic.AddInt32(&announces, -1) + } + } + // Create a valid chain and an infinite junk chain + targetBlocks := hashLimit + 2*maxQueueDist + hashes, blocks := makeChain(targetBlocks, 0, genesis) + validHeaderFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) + validBodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) + + attack, _ := makeChain(targetBlocks, 0, unknownBlock) + attackerHeaderFetcher := tester.makeHeaderFetcher("attacker", nil, -gatherSlack) + attackerBodyFetcher := tester.makeBodyFetcher("attacker", nil, 0) + + // 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], uint64(i+1), time.Now(), validHeaderFetcher, validBodyFetcher) + } + tester.fetcher.Notify("attacker", attack[i], 1 /* don't distance drop */, time.Now(), attackerHeaderFetcher, attackerBodyFetcher) + } + if count := atomic.LoadInt32(&announces); count != hashLimit+maxQueueDist { + t.Fatalf("queued announce count mismatch: have %d, want %d", count, hashLimit+maxQueueDist) + } + // Wait for fetches to complete + verifyImportCount(t, imported, maxQueueDist) + + // 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], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), validHeaderFetcher, validBodyFetcher) + verifyImportEvent(t, imported, true) + } + verifyImportDone(t, imported) +} + +// Tests that blocks sent to the fetcher (either through propagation or via hash +// announces and retrievals) don't pile up indefinitely, exhausting available +// system memory. +func TestBlockMemoryExhaustionAttack(t *testing.T) { + // Create a tester with instrumented import hooks + tester := newTester() + + imported, enqueued := make(chan *types.Block), int32(0) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + tester.fetcher.queueChangeHook = func(hash common.Hash, added bool) { + if added { + atomic.AddInt32(&enqueued, 1) + } else { + atomic.AddInt32(&enqueued, -1) + } + } + // Create a valid chain and a batch of dangling (but in range) blocks + targetBlocks := hashLimit + 2*maxQueueDist + hashes, blocks := makeChain(targetBlocks, 0, genesis) + attack := make(map[common.Hash]*types.Block) + for i := byte(0); len(attack) < blockLimit+2*maxQueueDist; i++ { + hashes, blocks := makeChain(maxQueueDist-1, i, unknownBlock) + for _, hash := range hashes[:maxQueueDist-2] { + attack[hash] = blocks[hash] + } + } + // Try to feed all the attacker blocks make sure only a limited batch is accepted + for _, block := range attack { + tester.fetcher.Enqueue("attacker", block) + } + time.Sleep(200 * time.Millisecond) + if queued := atomic.LoadInt32(&enqueued); queued != blockLimit { + t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit) + } + // Queue up a batch of valid blocks, and check that a new peer is allowed to do so + for i := 0; i < maxQueueDist-1; i++ { + tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-3-i]]) + } + time.Sleep(100 * time.Millisecond) + if queued := atomic.LoadInt32(&enqueued); queued != blockLimit+maxQueueDist-1 { + t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit+maxQueueDist-1) + } + // Insert the missing piece (and sanity check the import) + tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2]]) + verifyImportCount(t, imported, maxQueueDist) + + // Insert the remaining blocks in chunks to ensure clean DOS protection + for i := maxQueueDist; i < len(hashes)-1; i++ { + tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2-i]]) + verifyImportEvent(t, imported, true) + } + verifyImportDone(t, imported) +} diff --git a/dex/fetcher/metrics.go b/dex/fetcher/metrics.go new file mode 100644 index 000000000..23b670549 --- /dev/null +++ b/dex/fetcher/metrics.go @@ -0,0 +1,43 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Contains the metrics collected by the fetcher. + +package fetcher + +import ( + "github.com/dexon-foundation/dexon/metrics" +) + +var ( + propAnnounceInMeter = metrics.NewRegisteredMeter("dex/fetcher/prop/announces/in", nil) + propAnnounceOutTimer = metrics.NewRegisteredTimer("dex/fetcher/prop/announces/out", nil) + propAnnounceDropMeter = metrics.NewRegisteredMeter("dex/fetcher/prop/announces/drop", nil) + propAnnounceDOSMeter = metrics.NewRegisteredMeter("dex/fetcher/prop/announces/dos", nil) + + propBroadcastInMeter = metrics.NewRegisteredMeter("dex/fetcher/prop/broadcasts/in", nil) + propBroadcastOutTimer = metrics.NewRegisteredTimer("dex/fetcher/prop/broadcasts/out", nil) + propBroadcastDropMeter = metrics.NewRegisteredMeter("dex/fetcher/prop/broadcasts/drop", nil) + propBroadcastDOSMeter = metrics.NewRegisteredMeter("dex/fetcher/prop/broadcasts/dos", nil) + + headerFetchMeter = metrics.NewRegisteredMeter("dex/fetcher/fetch/headers", nil) + bodyFetchMeter = metrics.NewRegisteredMeter("dex/fetcher/fetch/bodies", nil) + + headerFilterInMeter = metrics.NewRegisteredMeter("dex/fetcher/filter/headers/in", nil) + headerFilterOutMeter = metrics.NewRegisteredMeter("dex/fetcher/filter/headers/out", nil) + bodyFilterInMeter = metrics.NewRegisteredMeter("dex/fetcher/filter/bodies/in", nil) + bodyFilterOutMeter = metrics.NewRegisteredMeter("dex/fetcher/filter/bodies/out", nil) +) -- cgit v1.2.3