Downloader.fetchHeadersPoW¶
| Source (upstream pre-purge) | Current | |
|---|---|---|
| File | eth/downloader/downloader.go |
downloader_pow.go |
| Symbol | Downloader.fetchHeaders |
Downloader.fetchHeadersPoW |
| Ref | 45baf2111~1 |
etc/v1.17.3-full-node |
Renamed from fetchHeaders. Diffs: LightSync head branch removed (#29711)
3-way merge — purge → getc ← upstream¶
pre-purge≈ adapted (origin inferred by similarity)fork-only
// fetchHeadersPoW keeps retrieving headers concurrently from the number
↗// requested, until no more are returned, potentially throttling on the way. To
// facilitate concurrency but still protect against malicious nodes sending bad
// headers, we construct a header chain skeleton using the "origin" peer we are
// syncing with, and fill in the missing headers using anyone else. Headers from
// other peers are only accepted if they map cleanly to the skeleton. If no one
// can fill in the skeleton - not even the origin peer - it's assumed invalid and
// the origin is dropped.
func (d *Downloader) fetchHeadersPoW(p *peerConnection, from uint64, head uint64, qwh *queueWithHeaders) error {
↗ p.log.Debug("Directing header downloads", "origin", from)
defer p.log.Debug("Header download terminated")
↗ // Start pulling the header chain skeleton until all is done
var (
skeleton = true // Skeleton assembly phase or finishing up
pivoting = false // Whether the next request is pivot verification
ancestor = from
↗ )
for {
// Pull the next batch of headers, it either:
↗ // - Skeleton retrieval to permit concurrent header fetches
// - Full header retrieval if we're near the chain head
var (
headers []*types.Header
hashes []common.Hash
err error
)
switch {
case pivoting:
d.pivotLock.RLock()
pivot := d.pivotHeader.Number.Uint64()
d.pivotLock.RUnlock()
↗ p.log.Trace("Fetching next pivot header", "number", pivot+uint64(fsMinFullBlocks))
headers, hashes, err = d.fetchHeadersByNumber(p, pivot+uint64(fsMinFullBlocks), 2, fsMinFullBlocks-9, false) // move +64 when it's 2x64-8 deep
↗ case skeleton:
p.log.Trace("Fetching skeleton headers", "count", MaxHeaderFetch, "from", from)
headers, hashes, err = d.fetchHeadersByNumber(p, from+uint64(MaxHeaderFetch)-1, MaxSkeletonSize, MaxHeaderFetch-1, false)
↗ default:
p.log.Trace("Fetching full headers", "count", MaxHeaderFetch, "from", from)
headers, hashes, err = d.fetchHeadersByNumber(p, from, MaxHeaderFetch, 0, false)
}
switch err {
case nil:
// Headers retrieved, continue with processing
↗ case errCanceled:
// Sync cancelled, no issue, propagate up
return err
↗ default:
// Header retrieval either timed out, or the peer failed in some strange way
// (e.g. disconnect). Consider the master peer bad and drop
d.dropPeer(p.id)
↗ // Finish the sync gracefully instead of dumping the gathered data though
for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} {
select {
case ch <- false:
case <-d.cancelCh:
}
}
select {
case d.headerProcCh <- nil:
case <-d.cancelCh:
}
return fmt.Errorf("%w: header request failed: %v", errBadPeer, err)
}
// If the pivot is being checked, move if it became stale and run the real retrieval
var pivot uint64
↗ d.pivotLock.RLock()
if d.pivotHeader != nil {
pivot = d.pivotHeader.Number.Uint64()
}
d.pivotLock.RUnlock()
↗ if pivoting {
if len(headers) == 2 {
if have, want := headers[0].Number.Uint64(), pivot+uint64(fsMinFullBlocks); have != want {
log.Warn("Peer sent invalid next pivot", "have", have, "want", want)
return fmt.Errorf("%w: next pivot number %d != requested %d", errInvalidChain, have, want)
}
if have, want := headers[1].Number.Uint64(), pivot+2*uint64(fsMinFullBlocks)-8; have != want {
log.Warn("Peer sent invalid pivot confirmer", "have", have, "want", want)
return fmt.Errorf("%w: next pivot confirmer number %d != requested %d", errInvalidChain, have, want)
}
log.Warn("Pivot seemingly stale, moving", "old", pivot, "new", headers[0].Number)
pivot = headers[0].Number.Uint64()
↗ d.pivotLock.Lock()
d.pivotHeader = headers[0]
d.pivotLock.Unlock()
↗ // Write out the pivot into the database so a rollback beyond
// it will reenable snap sync and update the state root that
// the state syncer will be downloading.
rawdb.WriteLastPivotNumber(d.stateDB, pivot)
}
// Disable the pivot check and fetch the next batch of headers
pivoting = false
continue
}
// If the skeleton's finished, pull any remaining head headers directly from the origin
if skeleton && len(headers) == 0 {
// A malicious node might withhold advertised headers indefinitely
if from+uint64(MaxHeaderFetch)-1 <= head {
p.log.Warn("Peer withheld skeleton headers", "advertised", head, "withheld", from+uint64(MaxHeaderFetch)-1)
return fmt.Errorf("%w: withheld skeleton headers: advertised %d, withheld #%d", errStallingPeer, head, from+uint64(MaxHeaderFetch)-1)
}
p.log.Debug("No skeleton, fetching headers directly")
skeleton = false
continue
}
// If no more headers are inbound, notify the content fetchers and return
if len(headers) == 0 {
// Don't abort header fetches while the pivot is downloading
if !d.committed.Load() && pivot <= from {
p.log.Debug("No headers, waiting for pivot commit")
select {
case <-time.After(fsHeaderContCheck):
continue
case <-d.cancelCh:
return errCanceled
}
}
// Pivot done (or not in snap sync) and no more headers, terminate the process
p.log.Debug("No more headers available")
select {
case d.headerProcCh <- nil:
return nil
case <-d.cancelCh:
return errCanceled
}
}
// If we received a skeleton batch, resolve internals concurrently
var progressed bool
if skeleton {
≈ filled, hashset, proced, err := d.fillHeaderSkeleton(from, headers, qwh)
↗ if err != nil {
p.log.Debug("Skeleton chain invalid", "err", err)
return fmt.Errorf("%w: %v", errInvalidChain, err)
}
headers = filled[proced:]
hashes = hashset[proced:]
↗ progressed = proced > 0
from += uint64(proced)
} else {
// A malicious node might withhold advertised headers indefinitely
if n := len(headers); n < MaxHeaderFetch && headers[n-1].Number.Uint64() < head {
p.log.Warn("Peer withheld headers", "advertised", head, "delivered", headers[n-1].Number.Uint64())
return fmt.Errorf("%w: withheld headers: advertised %d, delivered %d", errStallingPeer, head, headers[n-1].Number.Uint64())
}
// If we're closing in on the chain head, but haven't yet reached it, delay
// the last few headers so mini reorgs on the head don't cause invalid hash
// chain errors.
if n := len(headers); n > 0 {
// Retrieve the current head we're at
var head uint64
≈ head = d.blockchain.CurrentSnapBlock().Number.Uint64()
if full := d.blockchain.CurrentBlock().Number.Uint64(); head < full {
head = full
↗ }
// If the head is below the common ancestor, we're actually deduplicating
// already existing chain segments, so use the ancestor as the fake head.
// Otherwise, we might end up delaying header deliveries pointlessly.
if head < ancestor {
head = ancestor
}
// If the head is way older than this batch, delay the last few headers
if head+uint64(reorgProtThreshold) < headers[n-1].Number.Uint64() {
delay := reorgProtHeaderDelay
if delay > n {
delay = n
}
headers = headers[:n-delay]
hashes = hashes[:n-delay]
}
}
}
// If no headers have been delivered, or all of them have been delayed,
// sleep a bit and retry. Take care with headers already consumed during
// skeleton filling
if len(headers) == 0 && !progressed {
p.log.Trace("All headers delayed, waiting")
select {
case <-time.After(fsHeaderContCheck):
continue
case <-d.cancelCh:
return errCanceled
}
}
// Insert any remaining new headers and fetch the next batch
if len(headers) > 0 {
p.log.Trace("Scheduling new headers", "count", len(headers), "from", from)
select {
case d.headerProcCh <- &headerTask{
headers: headers,
hashes: hashes,
}:
case <-d.cancelCh:
return errCanceled
}
from += uint64(len(headers))
}
// If we're still skeleton filling snap sync, check pivot staleness
// before continuing to the next skeleton filling
if skeleton && pivot > 0 {
pivoting = true
}
}
}
core-geth validation — +3 -9
| | | |---|---| | File | [`downloader.go`](https://github.com/etclabscore/core-geth/blob/v1.12.20/eth/downloader/downloader.go) | | Symbol | `Downloader.fetchHeaders` | | Ref | `v1.12.20` |--- a/core-geth/eth/downloader/downloader.go
+++ b/etc/eth/downloader/downloader_pow.go
@@ -1,4 +1,4 @@
-// fetchHeaders keeps retrieving headers concurrently from the number
+// fetchHeadersPoW keeps retrieving headers concurrently from the number
// requested, until no more are returned, potentially throttling on the way. To
// facilitate concurrency but still protect against malicious nodes sending bad
// headers, we construct a header chain skeleton using the "origin" peer we are
@@ -6,7 +6,7 @@
// other peers are only accepted if they map cleanly to the skeleton. If no one
// can fill in the skeleton - not even the origin peer - it's assumed invalid and
// the origin is dropped.
-func (d *Downloader) fetchHeaders(p *peerConnection, from uint64, head uint64) error {
+func (d *Downloader) fetchHeadersPoW(p *peerConnection, from uint64, head uint64, qwh *queueWithHeaders) error {
p.log.Debug("Directing header downloads", "origin", from)
defer p.log.Debug("Header download terminated")
@@ -15,11 +15,9 @@
skeleton = true // Skeleton assembly phase or finishing up
pivoting = false // Whether the next request is pivot verification
ancestor = from
- mode = d.getMode()
)
for {
// Pull the next batch of headers, it either:
- // - Pivot check to see if the chain moved too far
// - Skeleton retrieval to permit concurrent header fetches
// - Full header retrieval if we're near the chain head
var (
@@ -140,7 +138,7 @@
// If we received a skeleton batch, resolve internals concurrently
var progressed bool
if skeleton {
- filled, hashset, proced, err := d.fillHeaderSkeleton(from, headers)
+ filled, hashset, proced, err := d.fillHeaderSkeleton(from, headers, qwh)
if err != nil {
p.log.Debug("Skeleton chain invalid", "err", err)
return fmt.Errorf("%w: %v", errInvalidChain, err)
@@ -162,14 +160,10 @@
if n := len(headers); n > 0 {
// Retrieve the current head we're at
var head uint64
- if mode == LightSync {
- head = d.lightchain.CurrentHeader().Number.Uint64()
- } else {
head = d.blockchain.CurrentSnapBlock().Number.Uint64()
if full := d.blockchain.CurrentBlock().Number.Uint64(); head < full {
head = full
}
- }
// If the head is below the common ancestor, we're actually deduplicating
// already existing chain segments, so use the ancestor as the fake head.
// Otherwise, we might end up delaying header deliveries pointlessly.