diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go index 084009a066..2b2f7d5fe0 100644 --- a/consensus/clique/clique.go +++ b/consensus/clique/clique.go @@ -213,14 +213,14 @@ func (c *Clique) Author(header *types.Header) (common.Address, error) { } // VerifyHeader checks whether a header conforms to the consensus rules. -func (c *Clique) VerifyHeader(chain consensus.ChainReader, header *types.Header, seal bool) error { +func (c *Clique) VerifyHeader(chain consensus.ChainHeaderReader, header *types.Header, seal bool) error { return c.verifyHeader(chain, header, nil) } // VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers. The // method returns a quit channel to abort the operations and a results channel to // retrieve the async verifications (the order is that of the input slice). -func (c *Clique) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { +func (c *Clique) VerifyHeaders(chain consensus.ChainHeaderReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { abort := make(chan struct{}) results := make(chan error, len(headers)) @@ -242,7 +242,7 @@ func (c *Clique) VerifyHeaders(chain consensus.ChainReader, headers []*types.Hea // caller may optionally pass in a batch of parents (ascending order) to avoid // looking those up from the database. This is useful for concurrently verifying // a batch of new headers. -func (c *Clique) verifyHeader(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error { +func (c *Clique) verifyHeader(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header) error { if header.Number == nil { return errUnknownBlock } @@ -305,7 +305,7 @@ func (c *Clique) verifyHeader(chain consensus.ChainReader, header *types.Header, // rather depend on a batch of previous headers. The caller may optionally pass // in a batch of parents (ascending order) to avoid looking those up from the // database. This is useful for concurrently verifying a batch of new headers. -func (c *Clique) verifyCascadingFields(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error { +func (c *Clique) verifyCascadingFields(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header) error { // The genesis block is the always valid dead-end number := header.Number.Uint64() if number == 0 { @@ -345,7 +345,7 @@ func (c *Clique) verifyCascadingFields(chain consensus.ChainReader, header *type } // snapshot retrieves the authorization snapshot at a given point in time. -func (c *Clique) snapshot(chain consensus.ChainReader, number uint64, hash common.Hash, parents []*types.Header) (*Snapshot, error) { +func (c *Clique) snapshot(chain consensus.ChainHeaderReader, number uint64, hash common.Hash, parents []*types.Header) (*Snapshot, error) { // Search for a snapshot in memory or on disk for checkpoints var ( headers []*types.Header @@ -436,7 +436,7 @@ func (c *Clique) VerifyUncles(chain consensus.ChainReader, block *types.Block) e // VerifySeal implements consensus.Engine, checking whether the signature contained // in the header satisfies the consensus protocol requirements. -func (c *Clique) VerifySeal(chain consensus.ChainReader, header *types.Header) error { +func (c *Clique) VerifySeal(chain consensus.ChainHeaderReader, header *types.Header) error { return c.verifySeal(chain, header, nil) } @@ -444,7 +444,7 @@ func (c *Clique) VerifySeal(chain consensus.ChainReader, header *types.Header) e // consensus protocol requirements. The method accepts an optional list of parent // headers that aren't yet part of the local blockchain to generate the snapshots // from. -func (c *Clique) verifySeal(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error { +func (c *Clique) verifySeal(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header) error { // Verifying the genesis block is not supported number := header.Number.Uint64() if number == 0 { @@ -654,7 +654,7 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, results c // CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty // that a new block should have based on the previous blocks in the chain and the // current signer. -func (c *Clique) CalcDifficulty(chain consensus.ChainReader, time uint64, parent *types.Header) *big.Int { +func (c *Clique) CalcDifficulty(chain consensus.ChainHeaderReader, time uint64, parent *types.Header) *big.Int { snap, err := c.snapshot(chain, parent.Number.Uint64(), parent.Hash(), nil) if err != nil { return nil diff --git a/consensus/consensus.go b/consensus/consensus.go index f753af550c..763f3c8ace 100644 --- a/consensus/consensus.go +++ b/consensus/consensus.go @@ -27,9 +27,9 @@ import ( "github.com/ethereum/go-ethereum/rpc" ) -// ChainReader defines a small collection of methods needed to access the local -// blockchain during header and/or uncle verification. -type ChainReader interface { +// ChainHeaderReader defines a small collection of methods needed to access the local +// blockchain during header verification. +type ChainHeaderReader interface { // Config retrieves the blockchain's chain configuration. Config() *params.ChainConfig @@ -44,6 +44,12 @@ type ChainReader interface { // GetHeaderByHash retrieves a block header from the database by its hash. GetHeaderByHash(hash common.Hash) *types.Header +} + +// ChainReader defines a small collection of methods needed to access the local +// blockchain during header and/or uncle verification. +type ChainReader interface { + ChainHeaderReader // GetBlock retrieves a block from the database by hash and number. GetBlock(hash common.Hash, number uint64) *types.Block @@ -59,13 +65,13 @@ type Engine interface { // VerifyHeader checks whether a header conforms to the consensus rules of a // given engine. Verifying the seal may be done optionally here, or explicitly // via the VerifySeal method. - VerifyHeader(chain ChainReader, header *types.Header, seal bool) error + VerifyHeader(chain ChainHeaderReader, header *types.Header, seal bool) error // VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers // concurrently. The method returns a quit channel to abort the operations and // a results channel to retrieve the async verifications (the order is that of // the input slice). - VerifyHeaders(chain ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) + VerifyHeaders(chain ChainHeaderReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) // VerifyUncles verifies that the given block's uncles conform to the consensus // rules of a given engine. @@ -73,7 +79,7 @@ type Engine interface { // VerifySeal checks whether the crypto seal on a header is valid according to // the consensus rules of the given engine. - VerifySeal(chain ChainReader, header *types.Header) error + VerifySeal(chain ChainHeaderReader, header *types.Header) error // Prepare initializes the consensus fields of a block header according to the // rules of a particular engine. The changes are executed inline. @@ -107,7 +113,7 @@ type Engine interface { // CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty // that a new block should have. - CalcDifficulty(chain ChainReader, time uint64, parent *types.Header) *big.Int + CalcDifficulty(chain ChainHeaderReader, time uint64, parent *types.Header) *big.Int // APIs returns the RPC APIs this consensus engine provides. APIs(chain ChainReader) []rpc.API diff --git a/consensus/ethash/consensus.go b/consensus/ethash/consensus.go index d271518f4f..3bb18f83b3 100644 --- a/consensus/ethash/consensus.go +++ b/consensus/ethash/consensus.go @@ -81,7 +81,7 @@ func (ethash *Ethash) Author(header *types.Header) (common.Address, error) { // VerifyHeader checks whether a header conforms to the consensus rules of the // stock Ethereum ethash engine. -func (ethash *Ethash) VerifyHeader(chain consensus.ChainReader, header *types.Header, seal bool) error { +func (ethash *Ethash) VerifyHeader(chain consensus.ChainHeaderReader, header *types.Header, seal bool) error { // If we're running a full engine faking, accept any input as valid if ethash.config.PowMode == ModeFullFake { return nil @@ -102,7 +102,7 @@ func (ethash *Ethash) VerifyHeader(chain consensus.ChainReader, header *types.He // VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers // concurrently. The method returns a quit channel to abort the operations and // a results channel to retrieve the async verifications. -func (ethash *Ethash) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { +func (ethash *Ethash) VerifyHeaders(chain consensus.ChainHeaderReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { // If we're running a full engine faking, accept any input as valid if ethash.config.PowMode == ModeFullFake || len(headers) == 0 { abort, results := make(chan struct{}), make(chan error, len(headers)) @@ -164,7 +164,7 @@ func (ethash *Ethash) VerifyHeaders(chain consensus.ChainReader, headers []*type return abort, errorsOut } -func (ethash *Ethash) verifyHeaderWorker(chain consensus.ChainReader, headers []*types.Header, seals []bool, index int) error { +func (ethash *Ethash) verifyHeaderWorker(chain consensus.ChainHeaderReader, headers []*types.Header, seals []bool, index int) error { var parent *types.Header if index == 0 { parent = chain.GetHeader(headers[0].ParentHash, headers[0].Number.Uint64()-1) @@ -238,7 +238,7 @@ func (ethash *Ethash) VerifyUncles(chain consensus.ChainReader, block *types.Blo // verifyHeader checks whether a header conforms to the consensus rules of the // stock Ethereum ethash engine. // See YP section 4.3.4. "Block Header Validity" -func (ethash *Ethash) verifyHeader(chain consensus.ChainReader, header, parent *types.Header, uncle bool, seal bool) error { +func (ethash *Ethash) verifyHeader(chain consensus.ChainHeaderReader, header, parent *types.Header, uncle bool, seal bool) error { // Ensure that the header's extra-data section is of a reasonable size if uint64(len(header.Extra)) > params.MaximumExtraDataSize { return fmt.Errorf("extra-data too long: %d > %d", len(header.Extra), params.MaximumExtraDataSize) @@ -301,7 +301,7 @@ func (ethash *Ethash) verifyHeader(chain consensus.ChainReader, header, parent * // CalcDifficulty is the difficulty adjustment algorithm. It returns // the difficulty that a new block should have when created at time // given the parent block's time and difficulty. -func (ethash *Ethash) CalcDifficulty(chain consensus.ChainReader, time uint64, parent *types.Header) *big.Int { +func (ethash *Ethash) CalcDifficulty(chain consensus.ChainHeaderReader, time uint64, parent *types.Header) *big.Int { return CalcDifficulty(chain.Config(), time, parent) } @@ -479,14 +479,14 @@ func calcDifficultyFrontier(time uint64, parent *types.Header) *big.Int { // VerifySeal implements consensus.Engine, checking whether the given block satisfies // the PoW difficulty requirements. -func (ethash *Ethash) VerifySeal(chain consensus.ChainReader, header *types.Header) error { +func (ethash *Ethash) VerifySeal(chain consensus.ChainHeaderReader, header *types.Header) error { return ethash.verifySeal(chain, header, false) } // verifySeal checks whether a block satisfies the PoW difficulty requirements, // either using the usual ethash cache for it, or alternatively using a full DAG // to make remote mining fast. -func (ethash *Ethash) verifySeal(chain consensus.ChainReader, header *types.Header, fulldag bool) error { +func (ethash *Ethash) verifySeal(chain consensus.ChainHeaderReader, header *types.Header, fulldag bool) error { // If we're running a fake PoW, accept any seal as valid if ethash.config.PowMode == ModeFake || ethash.config.PowMode == ModeFullFake { time.Sleep(ethash.fakeDelay) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 94f05f9674..c3b7718143 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -30,18 +30,21 @@ import ( ) const ( + lightTimeout = time.Millisecond // Time allowance before an announced header is explicitly requested 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 + hashLimit = 256 // Maximum number of unique blocks or headers a peer may have announced blockLimit = 64 // Maximum number of unique blocks a peer may have delivered + HeaderLimit = 64 // Maximum number of unique headers a peer may have delivered ) -var ( - errTerminated = errors.New("terminated") -) +var errTerminated = errors.New("terminated") + +// HeaderRetrievalFn is a callback type for retrieving a header from the local chain. +type HeaderRetrievalFn func(common.Hash) *types.Header // blockRetrievalFn is a callback type for retrieving a block from the local chain. type blockRetrievalFn func(common.Hash) *types.Block @@ -61,6 +64,9 @@ type blockBroadcasterFn func(block *types.Block, propagate bool) // chainHeightFn is a callback type to retrieve the current chain height. type chainHeightFn func() uint64 +// headersInsertFn is a callback type to insert a batch of headers into the local chain. +type headersInsertFn func(headers []*types.Header) (int, error) + // chainInsertFn is a callback type to insert a batch of blocks into the local chain. type chainInsertFn func(types.Blocks) (int, error) @@ -100,12 +106,32 @@ type bodyFilterTask struct { // inject represents a schedules import operation. type inject struct { origin string - block *types.Block + + header *types.Header // Used for light mode fetcher which only cares about header. + block *types.Block // Used for normal mode fetcher which imports full block. +} + +// number returns the block number of the injected object. +func (inject *inject) number() uint64 { + if inject.header != nil { + return inject.header.Number.Uint64() + } + return inject.block.NumberU64() +} + +// number returns the block hash of the injected object. +func (inject *inject) hash() common.Hash { + if inject.header != nil { + return inject.header.Hash() + } + return inject.block.Hash() } // Fetcher is responsible for accumulating block announcements from various peers // and scheduling them for retrieval. type Fetcher struct { + light bool // The indicator whether it's a light fetcher or not. + // Various event channels notify chan *announce inject chan *inject @@ -126,27 +152,30 @@ type Fetcher struct { // 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) + queued map[common.Hash]*inject // Set of already queued blocks (to dedup imports) // Callbacks + getHeader HeaderRetrievalFn // Retrieves a header from the local chain 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 + insertHeaders headersInsertFn // Injects a batch of headers into the chain 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) + 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.Header, *types.Block) // Method to call upon successful header or 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 { +func New(light bool, getHeader HeaderRetrievalFn, getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertHeaders headersInsertFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { return &Fetcher{ + light: light, notify: make(chan *announce), inject: make(chan *inject), headerFilter: make(chan chan *headerFilterTask), @@ -161,10 +190,12 @@ func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBloc queue: prque.New(nil), queues: make(map[string]int), queued: make(map[common.Hash]*inject), + getHeader: getHeader, getBlock: getBlock, verifyHeader: verifyHeader, broadcastBlock: broadcastBlock, chainHeight: chainHeight, + insertHeaders: insertHeaders, insertChain: insertChain, dropPeer: dropPeer, } @@ -202,8 +233,8 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time } } -// Enqueue tries to fill gaps the fetcher's future import queue. -func (f *Fetcher) Enqueue(peer string, block *types.Block) error { +// EnqueueBlock tries to fill gaps the fetcher's future import queue. +func (f *Fetcher) EnqueueBlock(peer string, block *types.Block) error { op := &inject{ origin: peer, block: block, @@ -290,12 +321,12 @@ func (f *Fetcher) loop() { height := f.chainHeight() for !f.queue.Empty() { op := f.queue.PopItem().(*inject) - hash := op.block.Hash() + hash := op.hash() if f.queueChangeHook != nil { f.queueChangeHook(hash, false) } // If too high up the chain or phase, continue later - number := op.block.NumberU64() + number := op.number() if number > height+1 { f.queue.Push(op, -int64(number)) if f.queueChangeHook != nil { @@ -304,11 +335,15 @@ func (f *Fetcher) loop() { break } // Otherwise if fresh and still unknown, try and import - if number+maxUncleDist < height || f.getBlock(hash) != nil { + if number+maxUncleDist < height || f.light && f.getHeader(hash) != nil || !f.light && f.getBlock(hash) != nil { f.forgetBlock(hash) continue } - f.insert(op.origin, op.block) + if f.light { + f.importHeaders(op.origin, op.header) + } else { + f.importBlocks(op.origin, op.block) + } } // Wait for an outside event to occur select { @@ -353,7 +388,10 @@ func (f *Fetcher) loop() { 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) + + if !f.light { + f.enqueueBlock(op.origin, op.block) + } case hash := <-f.done: // A pending import finished, remove all traces of the notification @@ -365,13 +403,18 @@ func (f *Fetcher) loop() { request := make(map[string][]common.Hash) for hash, announces := range f.announced { - if time.Since(announces[0].time) > arriveTimeout-gatherSlack { + timeout := arriveTimeout - gatherSlack + if f.light { + timeout = 0 + } + if time.Since(announces[0].time) > timeout { // 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 { + // If the block still didn't arrive or it's a light fetcher, + // queue for fetching. + if f.light && f.getHeader(hash) == nil || !f.light && f.getBlock(hash) == nil { request[announce.origin] = append(request[announce.origin], hash) f.fetching[hash] = announce } @@ -394,7 +437,9 @@ func (f *Fetcher) loop() { }() } // Schedule the next fetch if blocks are still pending - f.rescheduleFetch(fetchTimer) + if len(f.announced) >= 1 { + f.rescheduleFetch(fetchTimer) + } case <-completeTimer.C: // At least one header's timer ran out, retrieve everything @@ -423,7 +468,9 @@ func (f *Fetcher) loop() { go f.completing[hashes[0]].fetchBodies(hashes) } // Schedule the next fetch if blocks are still pending - f.rescheduleComplete(completeTimer) + if len(f.fetched) >= 1 { + f.rescheduleComplete(completeTimer) + } case filter := <-f.headerFilter: // Headers arrived from a remote peer. Extract those that were explicitly @@ -439,7 +486,7 @@ func (f *Fetcher) loop() { // 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{} + unknown, incomplete, complete, lightHeaders := []*types.Header{}, []*announce{}, []*types.Block{}, []*announce{} for _, header := range task.headers { hash := header.Hash() @@ -452,6 +499,16 @@ func (f *Fetcher) loop() { f.forgetHash(hash) continue } + // Collect all headers only if we are running in light + // mode and the headers are not imported by other means. + if f.light { + if f.getHeader(hash) == nil { + announce.header = header + lightHeaders = append(lightHeaders, announce) + f.forgetHash(hash) + } + continue + } // Only keep if not imported by other means if f.getBlock(hash) == nil { announce.header = header @@ -496,10 +553,14 @@ func (f *Fetcher) loop() { f.rescheduleComplete(completeTimer) } } + // Schedule the header for light fetcher import + for _, announce := range lightHeaders { + f.enqueueHeader(announce.origin, announce.header) + } // Schedule the header-only blocks for import for _, block := range complete { if announce := f.completing[block.Hash()]; announce != nil { - f.enqueue(announce.origin, block) + f.enqueueBlock(announce.origin, block) } } @@ -555,7 +616,7 @@ func (f *Fetcher) loop() { // Schedule the retrieved blocks for ordered import for _, block := range blocks { if announce := f.completing[block.Hash()]; announce != nil { - f.enqueue(announce.origin, block) + f.enqueueBlock(announce.origin, block) } } } @@ -568,6 +629,12 @@ func (f *Fetcher) rescheduleFetch(fetch *time.Timer) { if len(f.announced) == 0 { return } + // Schedule announcement retrieval quickly for light mode + // since server won't send any headers to client. + if f.light { + fetch.Reset(lightTimeout) + return + } // Otherwise find the earliest expiring announcement earliest := time.Now() for _, announces := range f.announced { @@ -594,9 +661,44 @@ func (f *Fetcher) rescheduleComplete(complete *time.Timer) { complete.Reset(gatherSlack - time.Since(earliest)) } -// enqueue schedules a new future import operation, if the block to be imported +// enqueueHeader schedules a new header import operation, if the header to be imported // has not yet been seen. -func (f *Fetcher) enqueue(peer string, block *types.Block) { +func (f *Fetcher) enqueueHeader(peer string, header *types.Header) { + hash := header.Hash() + // Ensure the peer isn't DOSing us + count := f.queues[peer] + 1 + if count > HeaderLimit { + log.Debug("Discarded propagated header, exceeded allowance", "peer", peer, "number", header.Number, "hash", hash, "limit", blockLimit) + propBroadcastDOSMeter.Mark(1) + f.forgetHash(hash) + return + } + // Discard any past or too distant blocks + if dist := int64(header.Number.Uint64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { + log.Debug("Discarded propagated header, too far away", "peer", peer, "number", header.Number.Uint64(), "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, + header: header, + } + f.queues[peer] = count + f.queued[hash] = op + f.queue.Push(op, -int64(header.Number.Uint64())) + if f.queueChangeHook != nil { + f.queueChangeHook(op.block.Hash(), true) + } + log.Debug("Queued propagated header", "peer", peer, "number", header.Number.Uint64(), "hash", hash, "queued", f.queue.Size()) + } +} + +// enqueueBlock schedules a new future import operation, if the block to be imported +// has not yet been seen. +func (f *Fetcher) enqueueBlock(peer string, block *types.Block) { hash := block.Hash() // Ensure the peer isn't DOSing us @@ -630,17 +732,52 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { } } -// 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() +func (f *Fetcher) importHeaders(peer string, header *types.Header) { + hash := header.Hash() + log.Debug("Importing propagated header", "peer", peer, "number", header.Number, "hash", 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.getHeader(header.ParentHash) + if parent == nil { + log.Debug("Unknown parent of propagated header", "peer", peer, "number", header.Number, "hash", hash, "parent", header.ParentHash) + return + } + // Quickly validate the header and propagate the block if it passes + switch err := f.verifyHeader(header); err { + case nil: + case consensus.ErrFutureBlock: + // Weird future block, don't fail, but neither propagate + + default: + // Something went very wrong, drop the peer + log.Debug("Propagated header verification failed", "peer", peer, "number", header.Number, "hash", hash, "err", err) + f.dropPeer(peer) + return + } + // Run the actual import and log any issues + if _, err := f.insertHeaders([]*types.Header{header}); err != nil { + log.Debug("Propagated header import failed", "peer", peer, "number", header.Number, "hash", hash, "err", err) + return + } + // Invoke the testing hook if needed + if f.importedHook != nil { + f.importedHook(header, nil) + } + }() +} + +// importBlocks 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) importBlocks(peer string, block *types.Block) { + hash := block.Hash() + 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 { @@ -674,11 +811,25 @@ func (f *Fetcher) insert(peer string, block *types.Block) { // Invoke the testing hook if needed if f.importedHook != nil { - f.importedHook(block) + f.importedHook(nil, block) } }() } +func (f *Fetcher) hasHash(hash common.Hash) { + if _, ok := f.fetching[hash]; ok { + } + if _, ok := f.completing[hash]; ok { + } + if _, ok := f.queued[hash]; ok { + } + if f.light { + + } else { + + } +} + // forgetHash removes all traces of a block announcement from the fetcher's // internal state. func (f *Fetcher) forgetHash(hash common.Hash) { diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 83172c5348..34f93b9712 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -78,26 +78,36 @@ func makeChain(n int, seed byte, parent *types.Block) ([]common.Hash, map[common 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 + hashes []common.Hash // Hash chain belonging to the tester + headers map[common.Hash]*types.Header // Headers 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 { +func newTester(light bool) *fetcherTester { tester := &fetcherTester{ - hashes: []common.Hash{genesis.Hash()}, - blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis}, - drops: make(map[string]bool), + hashes: []common.Hash{genesis.Hash()}, + headers: map[common.Hash]*types.Header{genesis.Hash(): genesis.Header()}, + 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 = New(light, tester.getHeader, tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertHeaders, tester.insertChain, tester.dropPeer) tester.fetcher.Start() return tester } +// getHeader retrieves a header from the tester's block chain. +func (f *fetcherTester) getHeader(hash common.Hash) *types.Header { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.headers[hash] +} + // getBlock retrieves a block from the tester's block chain. func (f *fetcherTester) getBlock(hash common.Hash) *types.Block { f.lock.RLock() @@ -120,9 +130,33 @@ func (f *fetcherTester) chainHeight() uint64 { f.lock.RLock() defer f.lock.RUnlock() + if f.fetcher.light { + return f.headers[f.hashes[len(f.hashes)-1]].Number.Uint64() + } return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() } +// insertChain injects a new headers into the simulated chain. +func (f *fetcherTester) insertHeaders(headers []*types.Header) (int, error) { + f.lock.Lock() + defer f.lock.Unlock() + + for i, header := range headers { + // Make sure the parent in known + if _, ok := f.headers[header.ParentHash]; !ok { + return i, errors.New("unknown parent") + } + // Discard any new blocks if the same height already exists + if header.Number.Uint64() <= f.headers[f.hashes[len(f.hashes)-1]].Number.Uint64() { + return i, nil + } + // Otherwise build our current chain + f.hashes = append(f.hashes, header.Hash()) + f.headers[header.Hash()] = header + } + return 0, nil +} + // insertChain injects a new blocks into the simulated chain. func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) { f.lock.Lock() @@ -233,7 +267,7 @@ func verifyCompletingEvent(t *testing.T, completing chan []common.Hash, arrive b } // verifyImportEvent verifies that one single event arrive on an import channel. -func verifyImportEvent(t *testing.T, imported chan *types.Block, arrive bool) { +func verifyImportEvent(t *testing.T, imported chan interface{}, arrive bool) { if arrive { select { case <-imported: @@ -251,7 +285,7 @@ func verifyImportEvent(t *testing.T, imported chan *types.Block, arrive bool) { // 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) { +func verifyImportCount(t *testing.T, imported chan interface{}, count int) { for i := 0; i < count; i++ { select { case <-imported: @@ -263,7 +297,7 @@ func verifyImportCount(t *testing.T, imported chan *types.Block, count int) { } // verifyImportDone verifies that no more events are arriving on an import channel. -func verifyImportDone(t *testing.T, imported chan *types.Block) { +func verifyImportDone(t *testing.T, imported chan interface{}) { select { case <-imported: t.Fatalf("extra block imported") @@ -271,45 +305,62 @@ func verifyImportDone(t *testing.T, imported chan *types.Block) { } } +// verifyChainHeight verifies the chain height is as expected. +func verifyChainHeight(t *testing.T, fetcher *fetcherTester, height uint64) { + if fetcher.chainHeight() != height { + t.Fatalf("chain height mismatch, got %d, want %d", fetcher.chainHeight(), height) + } +} + // 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 TestFullSequentialAnnouncements(t *testing.T) { testSequentialAnnouncements(t, false) } +func TestLightSequentialAnnouncements(t *testing.T) { testSequentialAnnouncements(t, true) } -func testSequentialAnnouncements(t *testing.T, protocol int) { +func testSequentialAnnouncements(t *testing.T, light bool) { // Create a chain of blocks to import targetBlocks := 4 * hashLimit hashes, blocks := makeChain(targetBlocks, 0, genesis) - tester := newTester() + tester := newTester(light) 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 } - + imported := make(chan interface{}) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { + if light { + if header == nil { + t.Fatalf("Fetcher try to import empty header") + } + imported <- header + } else { + if block == nil { + t.Fatalf("Fetcher try to import empty 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) + verifyChainHeight(t, tester, uint64(len(hashes)-1)) } // 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 TestFullConcurrentAnnouncements(t *testing.T) { testConcurrentAnnouncements(t, false) } +func TestLightConcurrentAnnouncements(t *testing.T) { testConcurrentAnnouncements(t, true) } -func testConcurrentAnnouncements(t *testing.T, protocol int) { +func testConcurrentAnnouncements(t *testing.T, light bool) { // 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() + tester := newTester(light) firstHeaderFetcher := tester.makeHeaderFetcher("first", blocks, -gatherSlack) firstBodyFetcher := tester.makeBodyFetcher("first", blocks, 0) secondHeaderFetcher := tester.makeHeaderFetcher("second", blocks, -gatherSlack) @@ -325,8 +376,20 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) { return secondHeaderFetcher(hash) } // Iteratively announce blocks until all are imported - imported := make(chan *types.Block) - tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + imported := make(chan interface{}) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { + if light { + if header == nil { + t.Fatalf("Fetcher try to import empty header") + } + imported <- header + } else { + if block == nil { + t.Fatalf("Fetcher try to import empty 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) @@ -340,31 +403,42 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) { if int(counter) != targetBlocks { t.Fatalf("retrieval count mismatch: have %v, want %v", counter, targetBlocks) } + verifyChainHeight(t, tester, uint64(len(hashes)-1)) } // 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 TestFullOverlappingAnnouncements(t *testing.T) { testOverlappingAnnouncements(t, false) } +func TestLightOverlappingAnnouncements(t *testing.T) { testOverlappingAnnouncements(t, true) } -func testOverlappingAnnouncements(t *testing.T, protocol int) { +func testOverlappingAnnouncements(t *testing.T, light bool) { // Create a chain of blocks to import targetBlocks := 4 * hashLimit hashes, blocks := makeChain(targetBlocks, 0, genesis) - tester := newTester() + tester := newTester(light) 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) + imported := make(chan interface{}, len(hashes)-1) for i := 0; i < overlap; i++ { imported <- nil } - tester.fetcher.importedHook = func(block *types.Block) { imported <- block } - + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { + if light { + if header == nil { + t.Fatalf("Fetcher try to import empty header") + } + imported <- header + } else { + if block == nil { + t.Fatalf("Fetcher try to import empty 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 { @@ -375,19 +449,19 @@ func testOverlappingAnnouncements(t *testing.T, protocol int) { } // Wait for all the imports to complete and check count verifyImportCount(t, imported, overlap) + verifyChainHeight(t, tester, uint64(len(hashes)-1)) } // 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 TestFullPendingDeduplication(t *testing.T) { testPendingDeduplication(t, false) } +func TestLightPendingDeduplication(t *testing.T) { testPendingDeduplication(t, true) } -func testPendingDeduplication(t *testing.T, protocol int) { +func testPendingDeduplication(t *testing.T, light bool) { // 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() + tester := newTester(light) headerFetcher := tester.makeHeaderFetcher("repeater", blocks, -gatherSlack) bodyFetcher := tester.makeBodyFetcher("repeater", blocks, 0) @@ -404,40 +478,57 @@ func testPendingDeduplication(t *testing.T, protocol int) { return nil } // Announce the same block many times until it's fetched (wait for any pending ops) - for tester.getBlock(hashes[0]) == nil { + checkNonExist := func() bool { + return tester.getBlock(hashes[0]) == nil + } + if light { + checkNonExist = func() bool { + return tester.getHeader(hashes[0]) == nil + } + } + for checkNonExist() { 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) } + verifyChainHeight(t, tester, 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 TestFullRandomArrivalImport(t *testing.T) { testRandomArrivalImport(t, false) } +func TestLightRandomArrivalImport(t *testing.T) { testRandomArrivalImport(t, true) } -func testRandomArrivalImport(t *testing.T, protocol int) { +func testRandomArrivalImport(t *testing.T, light bool) { // 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() + tester := newTester(light) 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 } + imported := make(chan interface{}, len(hashes)-1) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { + if light { + if header == nil { + t.Fatalf("Fetcher try to import empty header") + } + imported <- header + } else { + if block == nil { + t.Fatalf("Fetcher try to import empty block") + } + imported <- block + } + } for i := len(hashes) - 1; i >= 0; i-- { if i != skip { @@ -448,27 +539,26 @@ func testRandomArrivalImport(t *testing.T, protocol int) { // 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) + verifyChainHeight(t, tester, uint64(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 TestFullQueueGapFill(t *testing.T) { testQueueGapFill(t, false) } -func testQueueGapFill(t *testing.T, protocol int) { +func testQueueGapFill(t *testing.T, light bool) { // 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() + tester := newTester(light) 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 } + imported := make(chan interface{}, len(hashes)-1) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block } for i := len(hashes) - 1; i >= 0; i-- { if i != skip { @@ -477,22 +567,21 @@ func testQueueGapFill(t *testing.T, protocol int) { } } // Fill the missing block directly as if propagated - tester.fetcher.Enqueue("valid", blocks[hashes[skip]]) + tester.fetcher.EnqueueBlock("valid", blocks[hashes[skip]]) verifyImportCount(t, imported, len(hashes)-1) + verifyChainHeight(t, tester, uint64(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 TestFullImportDeduplication(t *testing.T) { testImportDeduplication(t, false) } -func testImportDeduplication(t *testing.T, protocol int) { +func testImportDeduplication(t *testing.T, light bool) { // 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() + tester := newTester(light) headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) @@ -503,20 +592,20 @@ func testImportDeduplication(t *testing.T, protocol int) { } // Instrument the fetching and imported events fetching := make(chan []common.Hash) - imported := make(chan *types.Block, len(hashes)-1) + imported := make(chan interface{}, len(hashes)-1) tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes } - tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + tester.fetcher.importedHook = func(header *types.Header, 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]]) + tester.fetcher.EnqueueBlock("valid", blocks[hashes[0]]) + tester.fetcher.EnqueueBlock("valid", blocks[hashes[0]]) + tester.fetcher.EnqueueBlock("valid", blocks[hashes[0]]) // Fill the missing block directly as if propagated, and check import uniqueness - tester.fetcher.Enqueue("valid", blocks[hashes[1]]) + tester.fetcher.EnqueueBlock("valid", blocks[hashes[1]]) verifyImportCount(t, imported, 2) if counter != 2 { @@ -526,7 +615,9 @@ func testImportDeduplication(t *testing.T, protocol int) { // 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) { +func TestFullDistantPropagationDiscarding(t *testing.T) { testDistantPropagationDiscarding(t, false) } + +func testDistantPropagationDiscarding(t *testing.T, light bool) { // Create a long chain to import and define the discard boundaries hashes, blocks := makeChain(3*maxQueueDist, 0, genesis) head := hashes[len(hashes)/2] @@ -534,7 +625,7 @@ func TestDistantPropagationDiscarding(t *testing.T) { 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 := newTester(light) tester.lock.Lock() tester.hashes = []common.Hash{head} @@ -542,13 +633,13 @@ func TestDistantPropagationDiscarding(t *testing.T) { tester.lock.Unlock() // Ensure that a block with a lower number than the threshold is discarded - tester.fetcher.Enqueue("lower", blocks[hashes[low]]) + tester.fetcher.EnqueueBlock("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]]) + tester.fetcher.EnqueueBlock("higher", blocks[hashes[high]]) time.Sleep(10 * time.Millisecond) if !tester.fetcher.queue.Empty() { t.Fatalf("fetcher queued future block") @@ -558,11 +649,10 @@ func TestDistantPropagationDiscarding(t *testing.T) { // 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 TestFullDistantAnnouncementDiscarding(t *testing.T) { testDistantAnnouncementDiscarding(t, false) } +func TestLightDistantAnnouncementDiscarding(t *testing.T) { testDistantAnnouncementDiscarding(t, true) } -func testDistantAnnouncementDiscarding(t *testing.T, protocol int) { +func testDistantAnnouncementDiscarding(t *testing.T, light bool) { // Create a long chain to import and define the discard boundaries hashes, blocks := makeChain(3*maxQueueDist, 0, genesis) head := hashes[len(hashes)/2] @@ -570,10 +660,11 @@ func testDistantAnnouncementDiscarding(t *testing.T, protocol int) { 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 := newTester(light) tester.lock.Lock() tester.hashes = []common.Hash{head} + tester.headers = map[common.Hash]*types.Header{head: blocks[head].Header()} tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} tester.lock.Unlock() @@ -601,20 +692,31 @@ func testDistantAnnouncementDiscarding(t *testing.T, protocol int) { // 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 TestFullInvalidNumberAnnouncement(t *testing.T) { testInvalidNumberAnnouncement(t, false) } +func TestLightInvalidNumberAnnouncement(t *testing.T) { testInvalidNumberAnnouncement(t, true) } -func testInvalidNumberAnnouncement(t *testing.T, protocol int) { +func testInvalidNumberAnnouncement(t *testing.T, light bool) { // Create a single block to import and check numbers against hashes, blocks := makeChain(1, 0, genesis) - tester := newTester() + tester := newTester(light) 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 } + imported := make(chan interface{}) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { + if light { + if header == nil { + t.Fatalf("Fetcher try to import empty header") + } + imported <- header + } else { + if block == nil { + t.Fatalf("Fetcher try to import empty 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) @@ -646,15 +748,13 @@ func testInvalidNumberAnnouncement(t *testing.T, protocol int) { // 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 TestFullEmptyBlockShortCircuit(t *testing.T) { testEmptyBlockShortCircuit(t, false) } -func testEmptyBlockShortCircuit(t *testing.T, protocol int) { +func testEmptyBlockShortCircuit(t *testing.T, light bool) { // Create a chain of blocks to import hashes, blocks := makeChain(32, 0, genesis) - tester := newTester() + tester := newTester(light) headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack) bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0) @@ -665,8 +765,20 @@ func testEmptyBlockShortCircuit(t *testing.T, protocol int) { 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 } + imported := make(chan interface{}) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { + if light { + if header == nil { + t.Fatalf("Fetcher try to import empty header") + } + imported <- header + } else { + if block == nil { + t.Fatalf("Fetcher try to import empty block") + } + imported <- block + } + } // Iteratively announce blocks until all are imported for i := len(hashes) - 2; i >= 0; i-- { @@ -687,16 +799,12 @@ func testEmptyBlockShortCircuit(t *testing.T, protocol int) { // 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) { +func TestFullHashMemoryExhaustionAttack(t *testing.T) { // Create a tester with instrumented import hooks - tester := newTester() + tester := newTester(false) - imported, announces := make(chan *types.Block), int32(0) - tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + imported, announces := make(chan interface{}), int32(0) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block } tester.fetcher.announceChangeHook = func(hash common.Hash, added bool) { if added { atomic.AddInt32(&announces, 1) @@ -740,10 +848,10 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) { // system memory. func TestBlockMemoryExhaustionAttack(t *testing.T) { // Create a tester with instrumented import hooks - tester := newTester() + tester := newTester(false) - imported, enqueued := make(chan *types.Block), int32(0) - tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + imported, enqueued := make(chan interface{}), int32(0) + tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block } tester.fetcher.queueChangeHook = func(hash common.Hash, added bool) { if added { atomic.AddInt32(&enqueued, 1) @@ -763,7 +871,7 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) { } // Try to feed all the attacker blocks make sure only a limited batch is accepted for _, block := range attack { - tester.fetcher.Enqueue("attacker", block) + tester.fetcher.EnqueueBlock("attacker", block) } time.Sleep(200 * time.Millisecond) if queued := atomic.LoadInt32(&enqueued); queued != blockLimit { @@ -771,19 +879,19 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) { } // 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]]) + tester.fetcher.EnqueueBlock("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]]) + tester.fetcher.EnqueueBlock("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]]) + tester.fetcher.EnqueueBlock("valid", blocks[hashes[len(hashes)-2-i]]) verifyImportEvent(t, imported, true) } verifyImportDone(t, imported) diff --git a/eth/handler.go b/eth/handler.go index 58add2eafc..c2abd24955 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -211,7 +211,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne } return n, err } - manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer) + manager.fetcher = fetcher.New(false, nil, blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, nil, inserter, manager.removePeer) return manager, nil } @@ -683,7 +683,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // Mark the peer as owning the block and schedule it for import p.MarkBlock(request.Block.Hash()) - pm.fetcher.Enqueue(p.id, request.Block) + pm.fetcher.EnqueueBlock(p.id, request.Block) // Assuming the block is importable by the peer, but possibly not yet done so, // calculate the head hash and TD that the peer truly must have. diff --git a/les/client.go b/les/client.go index 875c31cadc..aa7b9aea2d 100644 --- a/les/client.go +++ b/les/client.go @@ -226,6 +226,7 @@ func (s *LightEthereum) Start(srvr *p2p.Server) error { // Start bloom request workers. s.wg.Add(bloomServiceThreads) s.startBloomHandlers(params.BloomBitsBlocksClient) + s.handler.start() s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.config.NetworkId) diff --git a/les/client_handler.go b/les/client_handler.go index 2c033b69ff..4d16cc89fb 100644 --- a/les/client_handler.go +++ b/les/client_handler.go @@ -60,10 +60,14 @@ func newClientHandler(ulcConfig *eth.ULCConfig, backend *LightEthereum) *clientH return handler } +func (h *clientHandler) start() { + h.fetcher.start() +} + func (h *clientHandler) stop() { close(h.closeCh) h.downloader.Terminate() - h.fetcher.close() + h.fetcher.stop() h.wg.Wait() } @@ -172,6 +176,9 @@ func (h *clientHandler) handleMsg(p *serverPeer) error { p.Log().Trace("Valid announcement signature") } p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth) + + // Update peer head information first + p.updateHead(req.Hash, req.Number, req.Td) h.fetcher.announce(p, &req) } case BlockHeadersMsg: @@ -183,11 +190,16 @@ func (h *clientHandler) handleMsg(p *serverPeer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } + headers := resp.Headers p.fcServer.ReceivedReply(resp.ReqID, resp.BV) - if h.fetcher.requestedID(resp.ReqID) { - h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers) - } else { - if err := h.downloader.DeliverHeaders(p.id, resp.Headers); err != nil { + + // Filter out any explicitly requested headers, deliver the rest to the downloader + filter := len(headers) == 1 + if filter { + headers = h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers) + } + if len(headers) != 0 || !filter { + if err := h.downloader.DeliverHeaders(p.id, headers); err != nil { log.Debug("Failed to deliver headers", "err", err) } } diff --git a/les/fetcher.go b/les/fetcher.go index 38a72e2030..5ed88a2aa3 100644 --- a/les/fetcher.go +++ b/les/fetcher.go @@ -17,884 +17,556 @@ package les import ( + "errors" "math/big" + "math/rand" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" - "github.com/ethereum/go-ethereum/consensus" + "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" ) const ( - blockDelayTimeout = time.Second * 10 // timeout for a peer to announce a head that has already been confirmed by others - maxNodeCount = 20 // maximum number of fetcherTreeNode entries remembered for each peer - serverStateAvailable = 100 // number of recent blocks where state availability is assumed + blockDelayTimeout = time.Second * 10 // timeout for a peer to announce a head that has already been confirmed by others ) +var ( + errUnknownPeer = errors.New("announcement from unknown peer") + errInvalidAnnouncement = errors.New("received announcement is not strictly monotonic") + errUselessAnnouncement = errors.New("received announcement is useless") +) + +// announce represents a new block announcement from server. +type announce struct { + data *announceData + peer *serverPeer + errCh chan error +} + +// request represent a record when the request is sent. +type request struct { + reqID uint64 + peer *serverPeer + sendAt time.Time +} + +// response represents a response packet from network as well as a channel +// to return all un-requested data. +type response struct { + reqID uint64 + headers []*types.Header + peer *serverPeer + remain chan []*types.Header +} + +// query represents an operation to query whether a peer has announced +// a specified hash. +type query struct { + hash common.Hash + peer *serverPeer + resCh chan bool +} + +// confirm items form a linked list that is expanded with a new item every time +// a new head with a higher Td than the previous one has been downloaded and +// validated. +// +// The list contains a series of maximum confirmed Td values and the time these +// values have been confirmed, both increasing monotonically. +// +// A maximum confirmed Td is calculated both globally for all peers and also for +// each individual peer (meaning that the given peer has announced the head and +// it has also been downloaded from any peer, either before or after the given +// announcement). +// +// The linked list has a global tail where new confirmed Td entries are added and a +// separate head for each peer, pointing to the next Td entry that is higher than +// the peer's max confirmed Td (nil if it has already confirmed the current global head). +type confirm struct { + time mclock.AbsTime + td *big.Int + next *confirm +} + // lightFetcher implements retrieval of newly announced headers. It also provides a peerHasBlock function for the // ODR system to ensure that we only request data related to a certain block from peers who have already processed // and announced that block. type lightFetcher struct { handler *clientHandler - chain *light.LightChain + chain *light.LightChain // The local light chain which maintains the canonical header chain. + fetcher *fetcher.Fetcher // The underlying fetcher which takes care block header retrieval. - lock sync.Mutex // lock protects access to the fetcher's internal state variables except sent requests - maxConfirmedTd *big.Int - peers map[*serverPeer]*fetcherPeerInfo - lastUpdateStats *updateStatsEntry - syncing bool - syncDone chan *serverPeer - - reqMu sync.RWMutex // reqMu protects access to sent header fetch requests - requested map[uint64]fetchRequest - deliverChn chan fetchResponse - timeoutChn chan uint64 - requestTriggered bool - requestTrigger chan struct{} - lastTrustedHeader *types.Header + // Channels + addPeer chan *serverPeer + delPeer chan *serverPeer + announceCh chan *announce + requestCh chan *request + timeoutCh chan uint64 + deliverCh chan *response + queryCh chan *query + syncCh chan struct{} + syncDone chan *types.Header closeCh chan struct{} wg sync.WaitGroup + + // Test fields or hooks + ignoreAnnounce bool + announceHook func(*serverPeer, *announceData) + newHeadHook func(*types.Header) + syncingHook func() + addDelayHook func(p *serverPeer, delay time.Duration) } -// fetcherPeerInfo holds fetcher-specific information about each active peer -type fetcherPeerInfo struct { - root, lastAnnounced *fetcherTreeNode - nodeCnt int - confirmedTd *big.Int - bestConfirmed *fetcherTreeNode - nodeByHash map[common.Hash]*fetcherTreeNode - firstUpdateStats *updateStatsEntry -} - -// fetcherTreeNode is a node of a tree that holds information about blocks recently -// announced and confirmed by a certain peer. Each new announce message from a peer -// adds nodes to the tree, based on the previous announced head and the reorg depth. -// There are three possible states for a tree node: -// - announced: not downloaded (known) yet, but we know its head, number and td -// - intermediate: not known, hash and td are empty, they are filled out when it becomes known -// - known: both announced by this peer and downloaded (from any peer). -// This structure makes it possible to always know which peer has a certain block, -// which is necessary for selecting a suitable peer for ODR requests and also for -// canonizing new heads. It also helps to always download the minimum necessary -// amount of headers with a single request. -type fetcherTreeNode struct { - hash common.Hash - number uint64 - td *big.Int - known, requested bool - parent *fetcherTreeNode - children []*fetcherTreeNode -} - -// fetchRequest represents a header download request -type fetchRequest struct { - hash common.Hash - amount uint64 - peer *serverPeer - sent mclock.AbsTime - timeout bool -} - -// fetchResponse represents a header download response -type fetchResponse struct { - reqID uint64 - headers []*types.Header - peer *serverPeer +// fetcherPeer holds fetcher-specific information about each active peer +type fetcherPeer struct { + latest *announceData // The latest announcement packet. + confirmedTd *big.Int // The maximum total difficulty which confirmed by local chain and announced by peer. + confirms *confirm // The confirms list which shared by all peers and fetcher itself. + announces map[common.Hash]*announceData // All announcement data. } // newLightFetcher creates a new light fetcher func newLightFetcher(h *clientHandler) *lightFetcher { + chain := h.backend.blockchain + + // Construct the fetcher by offering all necessary callbacks + validator := func(header *types.Header) error { + // Disable seal verification explicitly if we are running in ulc mode. + return h.backend.engine.VerifyHeader(chain, header, !h.isULCEnabled()) + } + heighter := func() uint64 { return chain.CurrentHeader().Number.Uint64() } + dropper := func(id string) { h.backend.peers.unregister(id) } + inserter := func(headers []*types.Header) (int, error) { + // Disable PoW checking explicitly if we are running ulc mode. + checkFreq := 1 + if h.isULCEnabled() { + checkFreq = 0 + } + return chain.InsertHeaderChain(headers, checkFreq) + } f := &lightFetcher{ - handler: h, - chain: h.backend.blockchain, - peers: make(map[*serverPeer]*fetcherPeerInfo), - deliverChn: make(chan fetchResponse, 100), - requested: make(map[uint64]fetchRequest), - timeoutChn: make(chan uint64), - requestTrigger: make(chan struct{}, 1), - syncDone: make(chan *serverPeer), - closeCh: make(chan struct{}), - maxConfirmedTd: big.NewInt(0), + handler: h, + fetcher: fetcher.New(true, chain.GetHeaderByHash, nil, validator, nil, heighter, inserter, nil, dropper), + chain: h.backend.blockchain, + addPeer: make(chan *serverPeer), + delPeer: make(chan *serverPeer), + announceCh: make(chan *announce), + requestCh: make(chan *request), + timeoutCh: make(chan uint64), + deliverCh: make(chan *response), + queryCh: make(chan *query), + syncCh: make(chan struct{}), + syncDone: make(chan *types.Header), + closeCh: make(chan struct{}), } h.backend.peers.subscribe(f) - - f.wg.Add(1) - go f.syncLoop() return f } -func (f *lightFetcher) close() { - close(f.closeCh) +func (f *lightFetcher) start() { + f.wg.Add(1) + f.fetcher.Start() + go f.mainloop() } -// syncLoop is the main event loop of the light fetcher -func (f *lightFetcher) syncLoop() { +func (f *lightFetcher) stop() { + close(f.closeCh) + f.fetcher.Stop() + f.wg.Wait() +} + +// syncLoop is the main event loop of the light fetcher, which is responsible for +// * announcement maintenance(ulc) +// * block header retrieval +// * re-sync trigger +// * response delay and announcement delay statistic +func (f *lightFetcher) mainloop() { defer f.wg.Done() + + var ( + syncDist = uint64(1) // Interval used to trigger a light resync. + syncing bool // Indicator whether the client is syncing + + ulc = f.handler.isULCEnabled() + headCh = make(chan core.ChainHeadEvent, 100) + peers = make(map[*serverPeer]*fetcherPeer) + trusted = make(map[common.Hash][]*serverPeer) + trustedNumber = make(map[common.Hash]uint64) + fetching = make(map[uint64]*request) + + // Local status + localConfirm *confirm + localHead = f.chain.CurrentHeader() + localTd = f.chain.GetTd(localHead.Hash(), localHead.Number.Uint64()) + ) + sub := f.chain.SubscribeChainHeadEvent(headCh) + defer sub.Unsubscribe() + + // resetLocal updates the local status with given header. + resetLocal := func(header *types.Header) { + localHead = header + localTd = f.chain.GetTd(header.Hash(), header.Number.Uint64()) + + // All confirm records will be linked together and shared by all peers. + // In this way, we can judge whether the announcement from peer has + // hysteresis according to the maximum difficulty of each peer being confirmed. + if localConfirm == nil || localConfirm.td.Cmp(localTd) < 0 { + newConfirm := &confirm{time: mclock.Now(), td: localTd} + if localConfirm != nil { + localConfirm.next = newConfirm + } + localConfirm = newConfirm + } + } + // trustedHeader returns an indicator whether the header is regarded as + // trusted. If we are running in the ulc mode, only when we receive enough + // same announcement from trusted server, the header will be trusted. + trustedHeader := func(hash common.Hash) bool { + agreed := len(trusted[hash]) + return 100*agreed/len(f.handler.ulc.trustedKeys) >= f.handler.ulc.minTrustedFraction + } + // updateQoS drops stale confirm items and updates announcement delay statistic. + updateQoS := func(p *serverPeer, fp *fetcherPeer) { + now := mclock.Now() + // Track global confirm list if haven't. + if fp.confirms == nil { + fp.confirms = localConfirm + } + // Discard stale confirm items and feed server pool a delay statistic. + if fp.confirmedTd != nil { + for fp.confirms != nil && fp.confirms.td.Cmp(fp.confirmedTd) <= 0 { + if f.addDelayHook != nil { + f.addDelayHook(p, time.Duration(now-fp.confirms.time)) + } + f.handler.backend.serverPool.adjustBlockDelay(p.poolEntry, time.Duration(now-fp.confirms.time)) + p.Log().Debug("Add announcement delay", "time", common.PrettyDuration(time.Duration(now-fp.confirms.time))) + fp.confirms = fp.confirms.next + } + } + // Drop expired confirm items and feed server pool a "timeout" statistic. + for fp.confirms != nil && fp.confirms.time <= now-mclock.AbsTime(blockDelayTimeout) { + if f.addDelayHook != nil { + f.addDelayHook(p, blockDelayTimeout) + p.Log().Debug("Add announcement delay", "time", common.PrettyDuration(blockDelayTimeout)) + } + f.handler.backend.serverPool.adjustBlockDelay(p.poolEntry, blockDelayTimeout) + fp.confirms = fp.confirms.next + } + } + for { select { - case <-f.closeCh: - return - // request loop keeps running until no further requests are necessary or possible - case <-f.requestTrigger: - f.lock.Lock() - var ( - rq *distReq - reqID uint64 - syncing bool - ) - if !f.syncing { - rq, reqID, syncing = f.nextRequest() - } - f.requestTriggered = rq != nil - f.lock.Unlock() + case p := <-f.addPeer: + peers[p] = &fetcherPeer{announces: make(map[common.Hash]*announceData)} + log.Debug("Register peer", "id", p.id) - if rq != nil { - if _, ok := <-f.handler.backend.reqDist.queue(rq); ok { - if syncing { - f.lock.Lock() - f.syncing = true - f.lock.Unlock() - } else { - go func() { - time.Sleep(softRequestTimeout) - f.reqMu.Lock() - req, ok := f.requested[reqID] - if ok { - req.timeout = true - f.requested[reqID] = req - } - f.reqMu.Unlock() - // keep starting new requests while possible - f.requestTrigger <- struct{}{} - }() - } + case p := <-f.delPeer: + delete(peers, p) + log.Debug("Unregister peer", "id", p.id) + + case anno := <-f.announceCh: + p, data := anno.peer, anno.data + p.Log().Debug("Received new announcement", "number", data.Number, "hash", data.Hash, "reorg", data.ReorgDepth) + + fp, exist := peers[p] + if !exist { + p.Log().Debug("Announcement from unknown peer") + anno.errCh <- errUnknownPeer + continue + } + // announced tds should be strictly monotonic. + if fp.latest != nil && data.Td.Cmp(fp.latest.Td) <= 0 { + p.Log().Debug("Received non-monotonic td", "current", data.Td, "previous", fp.latest.Td) + anno.errCh <- errInvalidAnnouncement + continue + } + // filter out stale announcement if the td is less than local one. + if localTd != nil && data.Td.Cmp(localTd) <= 0 { + if f.chain.HasHeader(data.Hash, data.Number) { + fp.latest, fp.confirmedTd = data, data.Td + updateQoS(p, fp) + anno.errCh <- nil + p.Log().Debug("Received announcement is stale", "local", localTd, "received", data.Td) } else { - f.requestTrigger <- struct{}{} + anno.errCh <- errUselessAnnouncement + p.Log().Debug("Received announcement is useless", "local", localTd, "received", data.Td) + } + continue + } + fp.latest, fp.announces[data.Hash] = data, data + + if !ulc && !syncing { + if data.Number > localHead.Number.Uint64()+syncDist { + // Trigger light sync if the new announcement is not continuous + // with local chain. + p.Log().Debug("Trigger light sync", "local", localHead.Number, "localhash", localHead.Hash(), "remote", data.Number, "remotehash", data.Hash) + f.requestReSync(data.Hash) + } else { + p.Log().Debug("Trigger header retrieval", "number", data.Number, "hash", data.Hash) + f.fetcher.Notify(p.id, data.Hash, data.Number, time.Now(), f.requestHeaderByHash(p, data.Hash), nil) } } - case reqID := <-f.timeoutChn: - f.reqMu.Lock() - req, ok := f.requested[reqID] - if ok { - delete(f.requested, reqID) + if ulc && p.trusted { + // Keep collecting announcement from trusted server even we are syncing. + number, hash := data.Number, data.Hash + trusted[hash], trustedNumber[hash] = append(trusted[hash], p), data.Number + + // Notify underlying fetcher to retrieve header or trigger a resync if + // we have receive enough announcements from trusted server. + if trustedHeader(hash) && !syncing { + if number > localHead.Number.Uint64()+syncDist { + p.Log().Debug("Trigger trusted light sync", "local", localHead.Number, "localhash", localHead.Hash(), "remote", data.Number, "remotehash", data.Hash) + f.requestReSync(data.Hash) + } else { + p := trusted[hash][rand.Intn(len(trusted[hash]))] + p.Log().Debug("Trigger trusted header retrieval", "number", data.Number, "hash", data.Hash) + f.fetcher.Notify(p.id, hash, number, time.Now(), f.requestHeaderByHash(p, hash), nil) + } + } } - f.reqMu.Unlock() - if ok { - f.handler.backend.serverPool.adjustResponseTime(req.peer.poolEntry, time.Duration(mclock.Now()-req.sent), true) - req.peer.Log().Debug("Fetching data timed out hard") - go f.handler.removePeer(req.peer.id) + anno.errCh <- nil + + case req := <-f.requestCh: + fetching[req.reqID] = req // Tracking all in-flight requests for response latency statistic. + + case id := <-f.timeoutCh: + if req, exist := fetching[id]; exist { + log.Debug("request timeout", "peer", req.peer.id, "reqid", id) + delete(fetching, id) + f.handler.backend.serverPool.adjustResponseTime(req.peer.poolEntry, time.Since(req.sendAt), true) + go f.handler.backend.peers.unregister(req.peer.id) } - case resp := <-f.deliverChn: - f.reqMu.Lock() - req, ok := f.requested[resp.reqID] - if ok && req.peer != resp.peer { - ok = false + + case resp := <-f.deliverCh: + if req := fetching[resp.reqID]; req != nil { + // Feed response delay statistic for server pool. + delete(fetching, resp.reqID) + f.handler.backend.serverPool.adjustResponseTime(req.peer.poolEntry, time.Since(req.sendAt), false) + + resp.remain <- f.fetcher.FilterHeaders(resp.peer.id, resp.headers, time.Now()) + } else { + // Discard the entire packet no matter it's a timeout response or unexpected one. + resp.remain <- resp.headers } - if ok { - delete(f.requested, resp.reqID) + + case q := <-f.queryCh: + fp := peers[q.peer] + q.resCh <- fp != nil && fp.announces[q.hash] != nil + + case ev := <-headCh: + // Short circuit if we are still syncing. + if syncing { + continue } - f.reqMu.Unlock() - if ok { - f.handler.backend.serverPool.adjustResponseTime(req.peer.poolEntry, time.Duration(mclock.Now()-req.sent), req.timeout) + resetLocal(ev.Block.Header()) + number, hash := localHead.Number.Uint64(), localHead.Hash() + + // Clean stale announcements from trusted server. + if ulc { + for h, n := range trustedNumber { + if n <= number { + delete(trustedNumber, h) + delete(trusted, h) + } + } } - f.lock.Lock() - if !ok || !(f.syncing || f.processResponse(req, resp)) { - resp.peer.Log().Debug("Failed processing response") - go f.handler.removePeer(resp.peer.id) + for p, fp := range peers { + // Update the maximum confirmed td of peer if it already announced it. + if _, exist := fp.announces[hash]; exist { + fp.confirmedTd = localTd + } + updateQoS(p, fp) + // Delete all stale announcements. + for h, anno := range fp.announces { + if h == hash || anno.Number < number { + delete(fp.announces, h) + } + } } - f.lock.Unlock() - case p := <-f.syncDone: - f.lock.Lock() - p.Log().Debug("Done synchronising with peer") - f.checkSyncedHeaders(p) - f.syncing = false - f.lock.Unlock() - f.requestTrigger <- struct{}{} // f.requestTriggered is always true here + if f.newHeadHook != nil { + f.newHeadHook(localHead) + } + log.Debug("receive new head", "number", number, "hash", hash) + + case <-f.syncCh: + syncing = true // Mark the syncing as true only if we truly start syncing. + + case origin := <-f.syncDone: + syncing = false // Reset the status + + // Rewind all untrusted headers for ulc mode. + if ulc { + head := f.chain.CurrentHeader() + ancestor := rawdb.FindCommonAncestor(f.handler.backend.chainDb, origin, head) + if ancestor == nil { + // todo how should we handle this + } + var untrusted []common.Hash + for head.Number.Cmp(ancestor.Number) > 0 { + hash := head.Hash() + if trustedHeader(hash) { + break + } + untrusted = append(untrusted, hash) + head = f.chain.GetHeader(head.ParentHash, head.Number.Uint64()-1) + } + if len(untrusted) > 0 { + for i, j := 0, len(untrusted)-1; i < j; i, j = i+1, j-1 { + untrusted[i], untrusted[j] = untrusted[j], untrusted[i] + } + f.chain.Rollback(untrusted) + } + } + // Reset local status. + resetLocal(f.chain.CurrentHeader()) + log.Debug("light sync finished", "number", localHead.Number, "hash", localHead.Hash()) + + case <-f.closeCh: + return } } } // registerPeer adds a new peer to the fetcher's peer set func (f *lightFetcher) registerPeer(p *serverPeer) { - p.lock.Lock() - p.hasBlock = func(hash common.Hash, number uint64, hasState bool) bool { - return f.peerHasBlock(p, hash, number, hasState) + select { + case f.addPeer <- p: + case <-f.closeCh: } - p.lock.Unlock() - - f.lock.Lock() - defer f.lock.Unlock() - f.peers[p] = &fetcherPeerInfo{nodeByHash: make(map[common.Hash]*fetcherTreeNode)} } // unregisterPeer removes a new peer from the fetcher's peer set func (f *lightFetcher) unregisterPeer(p *serverPeer) { - p.lock.Lock() - p.hasBlock = nil - p.lock.Unlock() - - f.lock.Lock() - defer f.lock.Unlock() - - // check for potential timed out block delay statistics - f.checkUpdateStats(p, nil) - delete(f.peers, p) + select { + case f.delPeer <- p: + case <-f.closeCh: + } } -// announce processes a new announcement message received from a peer, adding new -// nodes to the peer's block tree and removing old nodes if necessary +// announce processes a new announcement message received from a peer. func (f *lightFetcher) announce(p *serverPeer, head *announceData) { - f.lock.Lock() - defer f.lock.Unlock() - p.Log().Debug("Received new announcement", "number", head.Number, "hash", head.Hash, "reorg", head.ReorgDepth) - - fp := f.peers[p] - if fp == nil { - p.Log().Debug("Announcement from unknown peer") + if f.ignoreAnnounce { return } - - if fp.lastAnnounced != nil && head.Td.Cmp(fp.lastAnnounced.td) <= 0 { - // announced tds should be strictly monotonic - p.Log().Debug("Received non-monotonic td", "current", head.Td, "previous", fp.lastAnnounced.td) - go f.handler.removePeer(p.id) + if f.announceHook != nil { + f.announceHook(p, head) + } + errCh := make(chan error, 1) + select { + case f.announceCh <- &announce{peer: p, data: head, errCh: errCh}: + case <-f.closeCh: return } - - n := fp.lastAnnounced - for i := uint64(0); i < head.ReorgDepth; i++ { - if n == nil { - break - } - n = n.parent - } - // n is now the reorg common ancestor, add a new branch of nodes - if n != nil && (head.Number >= n.number+maxNodeCount || head.Number <= n.number) { - // if announced head block height is lower or same as n or too far from it to add - // intermediate nodes then discard previous announcement info and trigger a resync - n = nil - fp.nodeCnt = 0 - fp.nodeByHash = make(map[common.Hash]*fetcherTreeNode) - } - // check if the node count is too high to add new nodes, discard oldest ones if necessary - if n != nil { - // n is now the reorg common ancestor, add a new branch of nodes - // check if the node count is too high to add new nodes - locked := false - for uint64(fp.nodeCnt)+head.Number-n.number > maxNodeCount && fp.root != nil { - if !locked { - f.chain.LockChain() - defer f.chain.UnlockChain() - locked = true - } - // if one of root's children is canonical, keep it, delete other branches and root itself - var newRoot *fetcherTreeNode - for i, nn := range fp.root.children { - if rawdb.ReadCanonicalHash(f.handler.backend.chainDb, nn.number) == nn.hash { - fp.root.children = append(fp.root.children[:i], fp.root.children[i+1:]...) - nn.parent = nil - newRoot = nn - break - } - } - fp.deleteNode(fp.root) - if n == fp.root { - n = newRoot - } - fp.root = newRoot - if newRoot == nil || !f.checkKnownNode(p, newRoot) { - fp.bestConfirmed = nil - fp.confirmedTd = nil - } - - if n == nil { - break - } - } - if n != nil { - for n.number < head.Number { - nn := &fetcherTreeNode{number: n.number + 1, parent: n} - n.children = append(n.children, nn) - n = nn - fp.nodeCnt++ - } - n.hash = head.Hash - n.td = head.Td - fp.nodeByHash[n.hash] = n - } - } - - if n == nil { - // could not find reorg common ancestor or had to delete entire tree, a new root and a resync is needed - if fp.root != nil { - fp.deleteNode(fp.root) - } - n = &fetcherTreeNode{hash: head.Hash, number: head.Number, td: head.Td} - fp.root = n - fp.nodeCnt++ - fp.nodeByHash[n.hash] = n - fp.bestConfirmed = nil - fp.confirmedTd = nil - } - - f.checkKnownNode(p, n) - p.lock.Lock() - p.headInfo = blockInfo{Number: head.Number, Hash: head.Hash, Td: head.Td} - fp.lastAnnounced = n - p.lock.Unlock() - f.checkUpdateStats(p, nil) - if !f.requestTriggered { - f.requestTriggered = true - f.requestTrigger <- struct{}{} + err := <-errCh + switch err { + case errInvalidAnnouncement, errUselessAnnouncement: + f.handler.backend.peers.unregister(p.id) + default: } } -// peerHasBlock returns true if we can assume the peer knows the given block -// based on its announcements -func (f *lightFetcher) peerHasBlock(p *serverPeer, hash common.Hash, number uint64, hasState bool) bool { - f.lock.Lock() - defer f.lock.Unlock() - - fp := f.peers[p] - if fp == nil || fp.root == nil { +// queryAnnounced checks whether the specified peer has announced a given +// hash announcement. +func (f *lightFetcher) queryAnnounced(peer *serverPeer, hash common.Hash) bool { + resCh := make(chan bool, 1) + select { + case f.queryCh <- &query{peer: peer, hash: hash, resCh: resCh}: + return <-resCh + case <-f.closeCh: return false } +} - if hasState { - if fp.lastAnnounced == nil || fp.lastAnnounced.number > number+serverStateAvailable { - return false +// trackRequest sends a reqID to main loop for in-flight request tracking. +func (f *lightFetcher) trackRequest(peer *serverPeer, id uint64) { + select { + case f.requestCh <- &request{reqID: id, peer: peer, sendAt: time.Now()}: + case <-f.closeCh: + } +} + +// requestHeaderByHash constructs a header retrieval request and sends it to +// local request distributor. +// Note, we rely on the underlying eth/fetcher to retrieve and validate the response, +// so that we have to obey the rule of eth/fetcher which only accepts the response +// from given peer. +func (f *lightFetcher) requestHeaderByHash(peer *serverPeer, hash common.Hash) func(common.Hash) error { + return func(hash common.Hash) error { + req := &distReq{ + getCost: func(dp distPeer) uint64 { return dp.(*serverPeer).getRequestCost(GetBlockHeadersMsg, 1) }, + canSend: func(dp distPeer) bool { return dp.(*serverPeer) == peer }, + request: func(dp distPeer) func() { + id := genReqID() + cost := peer.getRequestCost(GetBlockHeadersMsg, 1) + peer.fcServer.QueuedRequest(id, cost) + + f.trackRequest(peer, id) + go func() { + time.Sleep(hardRequestTimeout) + f.timeoutCh <- id + }() + return func() { peer.requestHeadersByHash(id, hash, 1, 0, false) } + }, } + // We have to spawn a go routine here is we need to send query + // to main loop, otherwise it will stuck the whole loop. + go func() { + <-f.handler.backend.reqDist.queue(req) + }() + return nil } - - if f.syncing { - // always return true when syncing - // false positives are acceptable, a more sophisticated condition can be implemented later - return true - } - - if number >= fp.root.number { - // it is recent enough that if it is known, is should be in the peer's block tree - return fp.nodeByHash[hash] != nil - } - f.chain.LockChain() - defer f.chain.UnlockChain() - // if it's older than the peer's block tree root but it's in the same canonical chain - // as the root, we can still be sure the peer knows it - // - // when syncing, just check if it is part of the known chain, there is nothing better we - // can do since we do not know the most recent block hash yet - return rawdb.ReadCanonicalHash(f.handler.backend.chainDb, fp.root.number) == fp.root.hash && rawdb.ReadCanonicalHash(f.handler.backend.chainDb, number) == hash } -// requestAmount calculates the amount of headers to be downloaded starting -// from a certain head backwards -func (f *lightFetcher) requestAmount(p *serverPeer, n *fetcherTreeNode) uint64 { - amount := uint64(0) - nn := n - for nn != nil && !f.checkKnownNode(p, nn) { - nn = nn.parent - amount++ - } - if nn == nil { - amount = n.number - } - return amount -} - -// requestedID tells if a certain reqID has been requested by the fetcher -func (f *lightFetcher) requestedID(reqID uint64) bool { - f.reqMu.RLock() - _, ok := f.requested[reqID] - f.reqMu.RUnlock() - return ok -} - -// nextRequest selects the peer and announced head to be requested next, amount -// to be downloaded starting from the head backwards is also returned -func (f *lightFetcher) nextRequest() (*distReq, uint64, bool) { - var ( - bestHash common.Hash - bestAmount uint64 - bestTd *big.Int - bestSyncing bool - ) - bestHash, bestAmount, bestTd, bestSyncing = f.findBestRequest() - - if bestTd == f.maxConfirmedTd { - return nil, 0, false - } - - var rq *distReq - reqID := genReqID() - if bestSyncing { - rq = f.newFetcherDistReqForSync(bestHash) - } else { - rq = f.newFetcherDistReq(bestHash, reqID, bestAmount) - } - return rq, reqID, bestSyncing -} - -// findBestRequest finds the best head to request that has been announced by but not yet requested from a known peer. -// It also returns the announced Td (which should be verified after fetching the head), -// the necessary amount to request and whether a downloader sync is necessary instead of a normal header request. -func (f *lightFetcher) findBestRequest() (bestHash common.Hash, bestAmount uint64, bestTd *big.Int, bestSyncing bool) { - bestTd = f.maxConfirmedTd - bestSyncing = false - - for p, fp := range f.peers { - for hash, n := range fp.nodeByHash { - if f.checkKnownNode(p, n) || n.requested { - continue - } - // if ulc mode is disabled, isTrustedHash returns true - amount := f.requestAmount(p, n) - if (bestTd == nil || n.td.Cmp(bestTd) > 0 || amount < bestAmount) && f.isTrustedHash(hash) { - bestHash = hash - bestTd = n.td - bestAmount = amount - bestSyncing = fp.bestConfirmed == nil || fp.root == nil || !f.checkKnownNode(p, fp.root) - } - } - } - return -} - -// isTrustedHash checks if the block can be trusted by the minimum trusted fraction. -func (f *lightFetcher) isTrustedHash(hash common.Hash) bool { - if !f.handler.isULCEnabled() { - return true - } - var numAgreed int - for p, fp := range f.peers { - if !p.trusted { - continue - } - if _, ok := fp.nodeByHash[hash]; !ok { - continue - } - numAgreed++ - } - return 100*numAgreed/len(f.handler.ulc.trustedKeys) >= f.handler.ulc.minTrustedFraction -} - -func (f *lightFetcher) newFetcherDistReqForSync(bestHash common.Hash) *distReq { - return &distReq{ - getCost: func(dp distPeer) uint64 { - return 0 - }, +// requestReSync constructs a re-sync request based on a given block hash. +func (f *lightFetcher) requestReSync(headHash common.Hash) { + req := &distReq{ + getCost: func(dp distPeer) uint64 { return 0 }, canSend: func(dp distPeer) bool { p := dp.(*serverPeer) - f.lock.Lock() - defer f.lock.Unlock() - if p.announceOnly { return false } - - fp := f.peers[p] - return fp != nil && fp.nodeByHash[bestHash] != nil + return f.queryAnnounced(p, headHash) }, request: func(dp distPeer) func() { - if f.handler.isULCEnabled() { - //keep last trusted header before sync - f.setLastTrustedHeader(f.chain.CurrentHeader()) - } go func() { p := dp.(*serverPeer) - p.Log().Debug("Synchronisation started") + origin := f.chain.CurrentHeader() + + f.syncCh <- struct{}{} // Mark the status of fetcher as syncing. + defer func() { + f.syncDone <- origin + }() + if f.syncingHook != nil { + f.syncingHook() + } f.handler.synchronise(p) - f.syncDone <- p }() return nil }, } -} - -// newFetcherDistReq creates a new request for the distributor. -func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bestAmount uint64) *distReq { - return &distReq{ - getCost: func(dp distPeer) uint64 { - p := dp.(*serverPeer) - return p.getRequestCost(GetBlockHeadersMsg, int(bestAmount)) - }, - canSend: func(dp distPeer) bool { - p := dp.(*serverPeer) - f.lock.Lock() - defer f.lock.Unlock() - - if p.announceOnly { - return false - } - - fp := f.peers[p] - if fp == nil { - return false - } - n := fp.nodeByHash[bestHash] - return n != nil && !n.requested - }, - request: func(dp distPeer) func() { - p := dp.(*serverPeer) - f.lock.Lock() - fp := f.peers[p] - if fp != nil { - n := fp.nodeByHash[bestHash] - if n != nil { - n.requested = true - } - } - f.lock.Unlock() - - cost := p.getRequestCost(GetBlockHeadersMsg, int(bestAmount)) - p.fcServer.QueuedRequest(reqID, cost) - f.reqMu.Lock() - f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()} - f.reqMu.Unlock() - go func() { - time.Sleep(hardRequestTimeout) - f.timeoutChn <- reqID - }() - return func() { p.requestHeadersByHash(reqID, bestHash, int(bestAmount), 0, true) } - }, - } + // We have to spawn a go routine here is we need to send query + // to main loop, otherwise it will stuck the whole loop. + go func() { + <-f.handler.backend.reqDist.queue(req) + }() } // deliverHeaders delivers header download request responses for processing -func (f *lightFetcher) deliverHeaders(peer *serverPeer, reqID uint64, headers []*types.Header) { - f.deliverChn <- fetchResponse{reqID: reqID, headers: headers, peer: peer} -} - -// processResponse processes header download request responses, returns true if successful -func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) bool { - if uint64(len(resp.headers)) != req.amount || resp.headers[0].Hash() != req.hash { - req.peer.Log().Debug("Response content mismatch", "requested", len(resp.headers), "reqfrom", resp.headers[0], "delivered", req.amount, "delfrom", req.hash) - return false - } - headers := make([]*types.Header, req.amount) - for i, header := range resp.headers { - headers[int(req.amount)-1-i] = header - } - - if _, err := f.chain.InsertHeaderChain(headers, 1); err != nil { - if err == consensus.ErrFutureBlock { - return true - } - log.Debug("Failed to insert header chain", "err", err) - return false - } - tds := make([]*big.Int, len(headers)) - for i, header := range headers { - td := f.chain.GetTd(header.Hash(), header.Number.Uint64()) - if td == nil { - log.Debug("Total difficulty not found for header", "index", i+1, "number", header.Number, "hash", header.Hash()) - return false - } - tds[i] = td - } - f.newHeaders(headers, tds) - return true -} - -// newHeaders updates the block trees of all active peers according to a newly -// downloaded and validated batch or headers -func (f *lightFetcher) newHeaders(headers []*types.Header, tds []*big.Int) { - var maxTd *big.Int - - for p, fp := range f.peers { - if !f.checkAnnouncedHeaders(fp, headers, tds) { - p.Log().Debug("Inconsistent announcement") - go f.handler.removePeer(p.id) - } - if fp.confirmedTd != nil && (maxTd == nil || maxTd.Cmp(fp.confirmedTd) > 0) { - maxTd = fp.confirmedTd - } - } - - if maxTd != nil { - f.updateMaxConfirmedTd(maxTd) - } -} - -// checkAnnouncedHeaders updates peer's block tree if necessary after validating -// a batch of headers. It searches for the latest header in the batch that has a -// matching tree node (if any), and if it has not been marked as known already, -// sets it and its parents to known (even those which are older than the currently -// validated ones). Return value shows if all hashes, numbers and Tds matched -// correctly to the announced values (otherwise the peer should be dropped). -func (f *lightFetcher) checkAnnouncedHeaders(fp *fetcherPeerInfo, headers []*types.Header, tds []*big.Int) bool { - var ( - n *fetcherTreeNode - header *types.Header - td *big.Int - ) - - for i := len(headers) - 1; ; i-- { - if i < 0 { - if n == nil { - // no more headers and nothing to match - return true - } - // we ran out of recently delivered headers but have not reached a node known by this peer yet, continue matching - hash, number := header.ParentHash, header.Number.Uint64()-1 - td = f.chain.GetTd(hash, number) - header = f.chain.GetHeader(hash, number) - if header == nil || td == nil { - log.Error("Missing parent of validated header", "hash", hash, "number", number) - return false - } - } else { - header = headers[i] - td = tds[i] - } - hash := header.Hash() - number := header.Number.Uint64() - if n == nil { - n = fp.nodeByHash[hash] - } - if n != nil { - if n.td == nil { - // node was unannounced - if nn := fp.nodeByHash[hash]; nn != nil { - // if there was already a node with the same hash, continue there and drop this one - nn.children = append(nn.children, n.children...) - n.children = nil - fp.deleteNode(n) - n = nn - } else { - n.hash = hash - n.td = td - fp.nodeByHash[hash] = n - } - } - // check if it matches the header - if n.hash != hash || n.number != number || n.td.Cmp(td) != 0 { - // peer has previously made an invalid announcement - return false - } - if n.known { - // we reached a known node that matched our expectations, return with success - return true - } - n.known = true - if fp.confirmedTd == nil || td.Cmp(fp.confirmedTd) > 0 { - fp.confirmedTd = td - fp.bestConfirmed = n - } - n = n.parent - if n == nil { - return true - } - } - } -} - -// checkSyncedHeaders updates peer's block tree after synchronisation by marking -// downloaded headers as known. If none of the announced headers are found after -// syncing, the peer is dropped. -func (f *lightFetcher) checkSyncedHeaders(p *serverPeer) { - fp := f.peers[p] - if fp == nil { - p.Log().Debug("Unknown peer to check sync headers") - return - } - - n := fp.lastAnnounced - var td *big.Int - - var h *types.Header - if f.handler.isULCEnabled() { - var unapprovedHashes []common.Hash - // Overwrite last announced for ULC mode - h, unapprovedHashes = f.lastTrustedTreeNode(p) - //rollback untrusted blocks - f.chain.Rollback(unapprovedHashes) - //overwrite to last trusted - n = fp.nodeByHash[h.Hash()] - } - - //find last valid block - for n != nil { - if td = f.chain.GetTd(n.hash, n.number); td != nil { - break - } - n = n.parent - } - - // Now n is the latest downloaded/approved header after syncing - if n == nil { - p.Log().Debug("Synchronisation failed") - go f.handler.removePeer(p.id) - return - } - header := f.chain.GetHeader(n.hash, n.number) - f.newHeaders([]*types.Header{header}, []*big.Int{td}) -} - -// lastTrustedTreeNode return last approved treeNode and a list of unapproved hashes -func (f *lightFetcher) lastTrustedTreeNode(p *serverPeer) (*types.Header, []common.Hash) { - unapprovedHashes := make([]common.Hash, 0) - current := f.chain.CurrentHeader() - - if f.lastTrustedHeader == nil { - return current, unapprovedHashes - } - - canonical := f.chain.CurrentHeader() - if canonical.Number.Uint64() > f.lastTrustedHeader.Number.Uint64() { - canonical = f.chain.GetHeaderByNumber(f.lastTrustedHeader.Number.Uint64()) - } - commonAncestor := rawdb.FindCommonAncestor(f.handler.backend.chainDb, canonical, f.lastTrustedHeader) - if commonAncestor == nil { - log.Error("Common ancestor of last trusted header and canonical header is nil", "canonical hash", canonical.Hash(), "trusted hash", f.lastTrustedHeader.Hash()) - return current, unapprovedHashes - } - - for current.Hash() == commonAncestor.Hash() { - if f.isTrustedHash(current.Hash()) { - break - } - unapprovedHashes = append(unapprovedHashes, current.Hash()) - current = f.chain.GetHeader(current.ParentHash, current.Number.Uint64()-1) - } - return current, unapprovedHashes -} - -func (f *lightFetcher) setLastTrustedHeader(h *types.Header) { - f.lock.Lock() - defer f.lock.Unlock() - f.lastTrustedHeader = h -} - -// checkKnownNode checks if a block tree node is known (downloaded and validated) -// If it was not known previously but found in the database, sets its known flag -func (f *lightFetcher) checkKnownNode(p *serverPeer, n *fetcherTreeNode) bool { - if n.known { - return true - } - td := f.chain.GetTd(n.hash, n.number) - if td == nil { - return false - } - header := f.chain.GetHeader(n.hash, n.number) - // check the availability of both header and td because reads are not protected by chain db mutex - // Note: returning false is always safe here - if header == nil { - return false - } - - fp := f.peers[p] - if fp == nil { - p.Log().Debug("Unknown peer to check known nodes") - return false - } - if !f.checkAnnouncedHeaders(fp, []*types.Header{header}, []*big.Int{td}) { - p.Log().Debug("Inconsistent announcement") - go f.handler.removePeer(p.id) - } - if fp.confirmedTd != nil { - f.updateMaxConfirmedTd(fp.confirmedTd) - } - return n.known -} - -// deleteNode deletes a node and its child subtrees from a peer's block tree -func (fp *fetcherPeerInfo) deleteNode(n *fetcherTreeNode) { - if n.parent != nil { - for i, nn := range n.parent.children { - if nn == n { - n.parent.children = append(n.parent.children[:i], n.parent.children[i+1:]...) - break - } - } - } - for { - if n.td != nil { - delete(fp.nodeByHash, n.hash) - } - fp.nodeCnt-- - if len(n.children) == 0 { - return - } - for i, nn := range n.children { - if i == 0 { - n = nn - } else { - fp.deleteNode(nn) - } - } - } -} - -// updateStatsEntry items form a linked list that is expanded with a new item every time a new head with a higher Td -// than the previous one has been downloaded and validated. The list contains a series of maximum confirmed Td values -// and the time these values have been confirmed, both increasing monotonically. A maximum confirmed Td is calculated -// both globally for all peers and also for each individual peer (meaning that the given peer has announced the head -// and it has also been downloaded from any peer, either before or after the given announcement). -// The linked list has a global tail where new confirmed Td entries are added and a separate head for each peer, -// pointing to the next Td entry that is higher than the peer's max confirmed Td (nil if it has already confirmed -// the current global head). -type updateStatsEntry struct { - time mclock.AbsTime - td *big.Int - next *updateStatsEntry -} - -// updateMaxConfirmedTd updates the block delay statistics of active peers. Whenever a new highest Td is confirmed, -// adds it to the end of a linked list together with the time it has been confirmed. Then checks which peers have -// already confirmed a head with the same or higher Td (which counts as zero block delay) and updates their statistics. -// Those who have not confirmed such a head by now will be updated by a subsequent checkUpdateStats call with a -// positive block delay value. -func (f *lightFetcher) updateMaxConfirmedTd(td *big.Int) { - if f.maxConfirmedTd == nil || td.Cmp(f.maxConfirmedTd) > 0 { - f.maxConfirmedTd = td - newEntry := &updateStatsEntry{ - time: mclock.Now(), - td: td, - } - if f.lastUpdateStats != nil { - f.lastUpdateStats.next = newEntry - } - - f.lastUpdateStats = newEntry - for p := range f.peers { - f.checkUpdateStats(p, newEntry) - } - } -} - -// checkUpdateStats checks those peers who have not confirmed a certain highest Td (or a larger one) by the time it -// has been confirmed by another peer. If they have confirmed such a head by now, their stats are updated with the -// block delay which is (this peer's confirmation time)-(first confirmation time). After blockDelayTimeout has passed, -// the stats are updated with blockDelayTimeout value. In either case, the confirmed or timed out updateStatsEntry -// items are removed from the head of the linked list. -// If a new entry has been added to the global tail, it is passed as a parameter here even though this function -// assumes that it has already been added, so that if the peer's list is empty (all heads confirmed, head is nil), -// it can set the new head to newEntry. -func (f *lightFetcher) checkUpdateStats(p *serverPeer, newEntry *updateStatsEntry) { - now := mclock.Now() - fp := f.peers[p] - if fp == nil { - p.Log().Debug("Unknown peer to check update stats") - return - } - - if newEntry != nil && fp.firstUpdateStats == nil { - fp.firstUpdateStats = newEntry - } - for fp.firstUpdateStats != nil && fp.firstUpdateStats.time <= now-mclock.AbsTime(blockDelayTimeout) { - f.handler.backend.serverPool.adjustBlockDelay(p.poolEntry, blockDelayTimeout) - fp.firstUpdateStats = fp.firstUpdateStats.next - } - if fp.confirmedTd != nil { - for fp.firstUpdateStats != nil && fp.firstUpdateStats.td.Cmp(fp.confirmedTd) <= 0 { - f.handler.backend.serverPool.adjustBlockDelay(p.poolEntry, time.Duration(now-fp.firstUpdateStats.time)) - fp.firstUpdateStats = fp.firstUpdateStats.next - } - } +func (f *lightFetcher) deliverHeaders(peer *serverPeer, reqID uint64, headers []*types.Header) []*types.Header { + remain := make(chan []*types.Header, 1) + select { + case f.deliverCh <- &response{reqID: reqID, headers: headers, peer: peer, remain: remain}: + case <-f.closeCh: + return nil + } + return <-remain } diff --git a/les/fetcher_test.go b/les/fetcher_test.go new file mode 100644 index 0000000000..0a2db950cc --- /dev/null +++ b/les/fetcher_test.go @@ -0,0 +1,284 @@ +// Copyright 2019 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 les + +import ( + "testing" + "time" + + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/eth" + "github.com/ethereum/go-ethereum/p2p/enode" +) + +// verifyImportEvent verifies that one single event arrive on an import channel. +func verifyImportEvent(t *testing.T, imported chan interface{}, 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(20 * time.Millisecond): + } + } +} + +// verifyImportDone verifies that no more events are arriving on an import channel. +func verifyImportDone(t *testing.T, imported chan interface{}) { + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): + } +} + +// verifyChainHeight verifies the chain height is as expected. +func verifyChainHeight(t *testing.T, fetcher *lightFetcher, height uint64) { + local := fetcher.chain.CurrentHeader().Number.Uint64() + if local != height { + t.Fatalf("chain height mismatch, got %d, want %d", local, height) + } +} + +func TestSequentialAnnouncementsLes2(t *testing.T) { testSequentialAnnouncements(t, 2) } +func TestSequentialAnnouncementsLes3(t *testing.T) { testSequentialAnnouncements(t, 3) } + +func testSequentialAnnouncements(t *testing.T, protocol int) { + s, c, teardown := newClientServerEnv(t, 4, protocol, nil, nil, false, false) + defer teardown() + + // Create connected peer pair. + c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync. + p1, err1, _, err2 := newTestPeerPair("peer", protocol, s.handler, c.handler) + select { + case <-time.After(time.Millisecond * 100): + case err := <-err1: + t.Fatalf("peer 1 handshake error: %v", err) + case err := <-err2: + t.Fatalf("peer 2 handshake error: %v", err) + } + c.handler.fetcher.ignoreAnnounce = false + + importCh := make(chan interface{}) + c.handler.fetcher.newHeadHook = func(header *types.Header) { + importCh <- header + } + for i := uint64(1); i <= s.backend.Blockchain().CurrentHeader().Number.Uint64(); i++ { + header := s.backend.Blockchain().GetHeaderByNumber(i) + hash, number := header.Hash(), header.Number.Uint64() + td := rawdb.ReadTd(s.db, hash, number) + + announce := announceData{hash, number, td, 0, nil} + if p1.cpeer.announceType == announceTypeSigned { + announce.sign(s.handler.server.privateKey) + } + p1.cpeer.sendAnnounce(announce) + verifyImportEvent(t, importCh, true) + } + verifyImportDone(t, importCh) + verifyChainHeight(t, c.handler.fetcher, 4) +} + +func TestGappedAnnouncementsLes2(t *testing.T) { testGappedAnnouncements(t, 2) } +func TestGappedAnnouncementsLes3(t *testing.T) { testGappedAnnouncements(t, 3) } + +func testGappedAnnouncements(t *testing.T, protocol int) { + s, c, teardown := newClientServerEnv(t, 4, protocol, nil, nil, false, false) + defer teardown() + + // Create connected peer pair. + c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync. + p1, err1, _, err2 := newTestPeerPair("peer", protocol, s.handler, c.handler) + select { + case <-time.After(time.Millisecond * 100): + case err := <-err1: + t.Fatalf("peer 1 handshake error: %v", err) + case err := <-err2: + t.Fatalf("peer 2 handshake error: %v", err) + } + c.handler.fetcher.ignoreAnnounce = false + + // Prepare announcement by latest header. + latest := s.backend.Blockchain().CurrentHeader() + hash, number := latest.Hash(), latest.Number.Uint64() + td := rawdb.ReadTd(s.db, hash, number) + + // Sign the announcement if necessary. + announce := announceData{hash, number, td, 0, nil} + if p1.cpeer.announceType == announceTypeSigned { + announce.sign(s.handler.server.privateKey) + } + p1.cpeer.sendAnnounce(announce) + time.Sleep(10 * time.Millisecond) + verifyChainHeight(t, c.handler.fetcher, 4) +} + +func TestTrustedAnnouncementsLes2(t *testing.T) { testTrustedAnnouncement(t, 2) } +func TestTrustedAnnouncementsLes3(t *testing.T) { testTrustedAnnouncement(t, 3) } + +func testTrustedAnnouncement(t *testing.T, protocol int) { + var ( + servers []*testServer + teardowns []func() + nodes []*enode.Node + ids []string + cpeers []*clientPeer + speers []*serverPeer + ) + for i := 0; i < 10; i++ { + s, n, teardown := newTestServerPeer(t, 10, protocol) + + servers = append(servers, s) + nodes = append(nodes, n) + teardowns = append(teardowns, teardown) + + // A half of them are trusted servers. + if i < 5 { + ids = append(ids, n.String()) + } + } + config := ð.ULCConfig{ + MinTrustedFraction: 60, // At least 3 approvals + TrustedServers: ids, + } + _, c, teardown := newClientServerEnv(t, 0, protocol, nil, config, false, false) + defer teardown() + defer func() { + for i := 0; i < len(teardowns); i++ { + teardowns[i]() + } + }() + + c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync. + + // Connect all server instances. + for i := 0; i < len(servers); i++ { + sp, cp, err := connect(servers[i].handler, nodes[i].ID(), c.handler, protocol) + if err != nil { + t.Fatalf("connect server and client failed, err %s", err) + } + cpeers = append(cpeers, cp) + speers = append(speers, sp) + } + c.handler.fetcher.ignoreAnnounce = false + + check := func(height []uint64, expected uint64, callback func()) { + for i := 0; i < len(height); i++ { + for j := 0; j < len(servers); j++ { + h := servers[j].backend.Blockchain().GetHeaderByNumber(height[i]) + hash, number := h.Hash(), h.Number.Uint64() + td := rawdb.ReadTd(servers[j].db, hash, number) + + // Sign the announcement if necessary. + announce := announceData{hash, number, td, 0, nil} + p := cpeers[j] + if p.announceType == announceTypeSigned { + announce.sign(servers[j].handler.server.privateKey) + } + p.sendAnnounce(announce) + + if j < 2 { + time.Sleep(10 * time.Millisecond) // Ensure the announcement has been processed. + if !c.handler.fetcher.queryAnnounced(speers[j], hash) { + t.Fatalf("the announcement from server peer %d should be kept", j+1) + } + } else if j == 2 { + // The block should be imported + } + } + } + if callback != nil { + callback() + } + time.Sleep(10 * time.Millisecond) // Ensure the announcement has been processed. + verifyChainHeight(t, c.handler.fetcher, expected) + } + check([]uint64{1}, 1, nil) // Sequential announcements + check([]uint64{4}, 4, nil) // ULC-style light syncing, rollback untrusted headers + + done := make(chan struct{}) + c.handler.fetcher.syncingHook = func() { <-done } + check([]uint64{6, 8}, 8, func() { done <- struct{}{} }) // ULC-style light syncing, keep the later trusted announces. + + c.handler.fetcher.syncingHook = nil + check([]uint64{10}, 10, nil) // Sync the whole chain. +} + +func TestAnnounceDelayLes2(t *testing.T) { testAnnounceDelay(t, 2) } +func TestAnnounceDelayLes3(t *testing.T) { testAnnounceDelay(t, 3) } + +func testAnnounceDelay(t *testing.T, protocol int) { + var ( + servers []*testServer + teardowns []func() + nodes []*enode.Node + cpeers []*clientPeer + speers []*serverPeer + ) + for i := 0; i < 5; i++ { + s, n, teardown := newTestServerPeer(t, 10, protocol) + + servers = append(servers, s) + nodes = append(nodes, n) + teardowns = append(teardowns, teardown) + } + _, c, teardown := newClientServerEnv(t, 0, protocol, nil, nil, false, false) + defer teardown() + defer func() { + for i := 0; i < len(teardowns); i++ { + teardowns[i]() + } + }() + + c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync. + // Connect all server instances. + for i := 0; i < len(servers); i++ { + sp, cp, err := connect(servers[i].handler, nodes[i].ID(), c.handler, protocol) + if err != nil { + t.Fatalf("connect server and client failed, err %s", err) + } + cpeers = append(cpeers, cp) + speers = append(speers, sp) + } + c.handler.fetcher.ignoreAnnounce = false + + delays := make(map[*serverPeer]time.Duration) + c.handler.fetcher.addDelayHook = func(p *serverPeer, delay time.Duration) { delays[p] = delay } + + for i := 0; i < 2; i++ { + h := servers[i].backend.Blockchain().GetHeaderByNumber(1) + hash, number := h.Hash(), h.Number.Uint64() + td := rawdb.ReadTd(servers[i].db, hash, number) + + announce := announceData{hash, number, td, 0, nil} + p := cpeers[i] + if p.announceType == announceTypeSigned { + announce.sign(servers[i].handler.server.privateKey) + } + p.sendAnnounce(announce) + } + time.Sleep(10 * time.Millisecond) + if d, exist := delays[speers[1]]; !exist || d > time.Millisecond { + t.Fatalf("the second announcement should be confirmed soon") + } +} diff --git a/les/odr_requests.go b/les/odr_requests.go index 146da2213c..f931ab4102 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -84,7 +84,7 @@ func (r *BlockRequest) GetCost(peer *serverPeer) uint64 { // CanSend tells if a certain peer is suitable for serving the given request func (r *BlockRequest) CanSend(peer *serverPeer) bool { - return peer.HasBlock(r.Hash, r.Number, false) + return peer.hasBlock(r.Hash, r.Number, false) } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) @@ -140,7 +140,7 @@ func (r *ReceiptsRequest) GetCost(peer *serverPeer) uint64 { // CanSend tells if a certain peer is suitable for serving the given request func (r *ReceiptsRequest) CanSend(peer *serverPeer) bool { - return peer.HasBlock(r.Hash, r.Number, false) + return peer.hasBlock(r.Hash, r.Number, false) } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) @@ -197,7 +197,7 @@ func (r *TrieRequest) GetCost(peer *serverPeer) uint64 { // CanSend tells if a certain peer is suitable for serving the given request func (r *TrieRequest) CanSend(peer *serverPeer) bool { - return peer.HasBlock(r.Id.BlockHash, r.Id.BlockNumber, true) + return peer.hasBlock(r.Id.BlockHash, r.Id.BlockNumber, true) } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) @@ -251,7 +251,7 @@ func (r *CodeRequest) GetCost(peer *serverPeer) uint64 { // CanSend tells if a certain peer is suitable for serving the given request func (r *CodeRequest) CanSend(peer *serverPeer) bool { - return peer.HasBlock(r.Id.BlockHash, r.Id.BlockNumber, true) + return peer.hasBlock(r.Id.BlockHash, r.Id.BlockNumber, true) } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) diff --git a/les/odr_test.go b/les/odr_test.go index f71f11b054..72ab033f7f 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -213,13 +213,13 @@ func testOdr(t *testing.T, protocol int, expFail uint64, checkCached bool, fn od // expect retrievals to fail (except genesis block) without a les peer client.handler.backend.peers.lock.Lock() - client.peer.speer.hasBlock = func(common.Hash, uint64, bool) bool { return false } + client.peer.speer.hasBlockCallback = func() bool { return false } client.handler.backend.peers.lock.Unlock() test(expFail) // expect all retrievals to pass client.handler.backend.peers.lock.Lock() - client.peer.speer.hasBlock = func(common.Hash, uint64, bool) bool { return true } + client.peer.speer.hasBlockCallback = nil client.handler.backend.peers.lock.Unlock() test(5) diff --git a/les/peer.go b/les/peer.go index 01fd20fee1..fa0cfd043c 100644 --- a/les/peer.go +++ b/les/peer.go @@ -311,8 +311,8 @@ type serverPeer struct { updateCount uint64 updateTime mclock.AbsTime - // Callbacks - hasBlock func(common.Hash, uint64, bool) bool // Used to determine whether the server has the specified block. + // Test Hooks + hasBlockCallback func() bool } func newServerPeer(version int, network uint64, trusted bool, p *p2p.Peer, rw p2p.MsgReadWriter) *serverPeer { @@ -475,11 +475,14 @@ func (p *serverPeer) getTxRelayCost(amount, size int) uint64 { return cost } -// HasBlock checks if the peer has a given block -func (p *serverPeer) HasBlock(hash common.Hash, number uint64, hasState bool) bool { +// hasBlock checks if the peer has a given block +func (p *serverPeer) hasBlock(hash common.Hash, number uint64, hasState bool) bool { p.lock.RLock() defer p.lock.RUnlock() + if p.hasBlockCallback != nil { + return p.hasBlockCallback() + } head := p.headInfo.Number var since, recent uint64 if hasState { @@ -489,9 +492,7 @@ func (p *serverPeer) HasBlock(hash common.Hash, number uint64, hasState bool) bo since = p.chainSince recent = p.chainRecent } - hasBlock := p.hasBlock - - return head >= number && number >= since && (recent == 0 || number+recent+4 > head) && hasBlock != nil && hasBlock(hash, number, hasState) + return head >= number && number >= since && (recent == 0 || number+recent+4 > head) } // updateFlowControl updates the flow control parameters belonging to the server @@ -516,6 +517,15 @@ func (p *serverPeer) updateFlowControl(update keyValueMap) { } } +// updateHead updates the head information based on the announcement from +// the peer. +func (p *serverPeer) updateHead(hash common.Hash, number uint64, td *big.Int) { + p.lock.Lock() + defer p.lock.Unlock() + + p.headInfo = blockInfo{Hash: hash, Number: number, Td: td} +} + // Handshake executes the les protocol handshake, negotiating version number, // network IDs, difficulties, head and genesis blocks. func (p *serverPeer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, server *LesServer) error { diff --git a/les/test_helper.go b/les/test_helper.go index 3681ff559c..761f1fc825 100644 --- a/les/test_helper.go +++ b/les/test_helper.go @@ -23,6 +23,7 @@ import ( "context" "crypto/rand" "math/big" + "net" "testing" "time" @@ -221,6 +222,7 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index if client.registrar != nil { client.registrar.start(backend) } + client.handler.start() return client.handler } @@ -280,6 +282,8 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da }, fcManager: flowcontrol.NewClientManager(nil, clock), } + key, _ := crypto.GenerateKey() + server.privateKey = key server.costTracker, server.minCapacity = newCostTracker(db, server.config, nil) server.costTracker.costListHook = func() RequestCostList { return testCostList(0) } // Disable flow control mechanism. server.handler = newServerHandler(server, simulation.Blockchain(), db, txpool, nil, func() bool { return true }) @@ -349,14 +353,13 @@ func (p *testPeer) close() { func newTestPeerPair(name string, version int, server *serverHandler, client *clientHandler) (*testPeer, <-chan error, *testPeer, <-chan error) { // Create a message pipe to communicate through - app, net := p2p.MsgPipe() + app, s := p2p.MsgPipe() // Generate a random id and create the peer - var id enode.ID - rand.Read(id[:]) + en := enode.NewV4(&server.server.privateKey.PublicKey, net.ParseIP("127.0.0.1"), 35000, 35000) - peer1 := newClientPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net) - peer2 := newServerPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), app) + peer1 := newClientPeer(version, NetworkId, p2p.NewPeer(en.ID(), name, nil), s) + peer2 := newServerPeer(version, NetworkId, false, p2p.NewPeer(en.ID(), name, nil), app) // Start the peer on a new thread errc1 := make(chan error, 1) @@ -375,7 +378,7 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl case errc1 <- client.handle(peer2): } }() - return &testPeer{cpeer: peer1, net: net, app: app}, errc1, &testPeer{speer: peer2, net: app, app: net}, errc2 + return &testPeer{cpeer: peer1, net: s, app: app}, errc1, &testPeer{speer: peer2, net: app, app: s}, errc2 } // handshake simulates a trivial handshake that expects the same state from the diff --git a/les/txrelay.go b/les/txrelay.go index 33e198fdc6..16ddfbc145 100644 --- a/les/txrelay.go +++ b/les/txrelay.go @@ -62,14 +62,18 @@ func (self *lesTxRelay) registerPeer(p *serverPeer) { self.lock.Lock() defer self.lock.Unlock() - self.peerList = self.ps.allServerPeers() + self.peerList = append(self.peerList, p) } func (self *lesTxRelay) unregisterPeer(p *serverPeer) { self.lock.Lock() defer self.lock.Unlock() - self.peerList = self.ps.allServerPeers() + for index, peer := range self.peerList { + if peer == p { + self.peerList = append(self.peerList[:index], self.peerList[index+1:]...) + } + } } // send sends a list of transactions to at most a given number of peers at diff --git a/light/lightchain.go b/light/lightchain.go index daaa4e1ad9..0306bcbed3 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -256,7 +256,7 @@ func (lc *LightChain) GetBodyRLP(ctx context.Context, hash common.Hash) (rlp.Raw return body, nil } -// HasBlock checks if a block is fully present in the database or not, caching +// hasBlock checks if a block is fully present in the database or not, caching // it if present. func (lc *LightChain) HasBlock(hash common.Hash, number uint64) bool { blk, _ := lc.GetBlock(NoOdr, hash, number)