Downloader.processHeadersPoW¶
| Source (upstream pre-purge) | Current | |
|---|---|---|
| File | eth/downloader/downloader.go |
downloader_pow.go |
| Symbol | Downloader.processHeaders |
Downloader.processHeadersPoW |
| Ref | 45baf2111~1 |
etc/v1.17.3-full-node |
Removed
beaconModeparam, TTD transition rejection, andLightSyncbranches (not applicable to ETC). Added upstream'schainCutoffNumber/InsertHeadersBeforeCutoffpath for ancient headers. TheInsertHeaderChaincall for snap sync headers is retained from the pre-purge version — ensures TDs are computed and persisted during header download.
3-way merge — purge → getc ← upstream¶
pre-purgeupstreamcommoncore-geth≈ adapted (origin inferred by similarity)fork-only
// processHeadersPoW takes batches of retrieved headers from an input channel and
// keeps processing and scheduling them into the header chain and downloader's
// queue until the stream ends or a failure occurs.
func (d *Downloader) processHeadersPoW(origin uint64, td *big.Int) error {
var (
↗ mode = d.getMode()
gotHeaders = false // Wait for batches of headers to process
timer = time.NewTimer(time.Second)
)
defer timer.Stop()
for {
select {
case <-d.cancelCh:
return errCanceled
case task := <-d.headerProcCh:
// Terminate header processing if we synced up
if task == nil || len(task.headers) == 0 {
// Notify everyone that headers are fully processed
for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} {
select {
case ch <- false:
case <-d.cancelCh:
}
}
// If no headers were retrieved at all, the peer violated its TD promise that it had a
// better chain compared to ours. The only exception is if its promised blocks were
// already imported by other means (e.g. fetcher):
//
// R <remote peer>, L <local node>: Both at block 10
// R: Mine block 11, and propagate it to L
// L: Queue block 11 for import
// L: Notice that R's head and TD increased compared to ours, start sync
// L: Import of block 11 finishes
// L: Sync begins, and finds common ancestor at 11
// L: Request new headers up from 11 (R's TD was higher, it must have something)
// R: Nothing to give
≈ head := d.blockchain.CurrentBlock()
if !gotHeaders && td.Cmp(d.blockchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
return errStallingPeer
↗ }
// If snap or light syncing, ensure promised headers are indeed delivered. This is
// needed to detect scenarios where an attacker feeds a bad pivot and then bails out
// of delivering the post-pivot blocks that would flag the invalid content.
//
// This check cannot be executed "as is" for full imports, since blocks may still be
// queued for processing when the header download completes. However, as long as the
// peer gave us something useful, we're already happy/progressed (above check).
≈ if mode == ethconfig.SnapSync {
≈ head := d.blockchain.CurrentHeader()
if td.Cmp(d.blockchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
return errStallingPeer
↗ }
}
return nil
}
// Otherwise split the chunk of headers into batches and process them
↗ headers, hashes, scheduled := task.headers, task.hashes, false
↗ gotHeaders = true
for len(headers) > 0 {
// Terminate if something failed in between processing chunks
select {
case <-d.cancelCh:
return errCanceled
default:
}
// Select the next chunk of headers to import
limit := maxHeadersProcess
if limit > len(headers) {
limit = len(headers)
}
chunkHeaders := headers[:limit]
chunkHashes := hashes[:limit]
↗ // Split the headers around the chain cutoff
var cutoff int
if mode == ethconfig.SnapSync && d.chainCutoffNumber != 0 {
cutoff = sort.Search(len(chunkHeaders), func(i int) bool {
return chunkHeaders[i].Number.Uint64() >= d.chainCutoffNumber
})
}
// Insert the header chain into the ancient store (with block bodies and
// receipts set to nil) if they fall before the cutoff.
if mode == ethconfig.SnapSync && cutoff != 0 {
if n, err := d.blockchain.InsertHeadersBeforeCutoff(chunkHeaders[:cutoff]); err != nil {
log.Warn("Failed to insert ancient header chain", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err)
return fmt.Errorf("%w: %v", errInvalidChain, err)
}
↗ log.Debug("Inserted headers before cutoff", "number", chunkHeaders[cutoff-1].Number, "hash", chunkHashes[cutoff-1])
}
// In snap sync, insert headers into the header chain so that TDs
// are computed and persisted. Without this, headers exist in the
// database but have no associated TD, breaking PoW sync operations
// that rely on TD comparisons (cf. core-geth processHeaders).
≈ if mode == ethconfig.SnapSync {
≈ if n, err := d.blockchain.InsertHeaderChain(chunkHeaders); err != nil {
log.Warn("Failed to insert header chain", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err)
return fmt.Errorf("%w: %v", errInvalidChain, err)
↗ }
}
↗ // If we've reached the allowed number of pending headers, stall a bit
for d.queue.PendingBodies() >= maxQueuedHeaders || d.queue.PendingReceipts() >= maxQueuedHeaders {
timer.Reset(time.Second)
select {
case <-d.cancelCh:
return errCanceled
case <-timer.C:
}
↗ }
// Otherwise, schedule the headers for content retrieval (block bodies and
// potentially receipts in snap sync).
//
// Skip the bodies/receipts retrieval scheduling before the cutoff in snap
// sync if chain pruning is configured.
if mode == ethconfig.SnapSync && cutoff != 0 {
chunkHeaders = chunkHeaders[cutoff:]
chunkHashes = chunkHashes[cutoff:]
}
if len(chunkHeaders) > 0 {
scheduled = true
if d.queue.Schedule(chunkHeaders, chunkHashes, origin+uint64(cutoff)) != len(chunkHeaders) {
return fmt.Errorf("%w: stale headers", errBadPeer)
}
}
headers = headers[limit:]
hashes = hashes[limit:]
origin += uint64(limit)
}
// Update the highest block number we know if a higher one is found.
d.syncStatsLock.Lock()
if d.syncStatsChainHeight < origin {
d.syncStatsChainHeight = origin - 1
}
d.syncStatsLock.Unlock()
↗ // Signal the content downloaders of the availability of new tasks
↗ if scheduled {
for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} {
select {
case ch <- true:
default:
}
}
}
}
}
}
core-geth validation — +47 -127
| | | |---|---| | File | [`downloader.go`](https://github.com/etclabscore/core-geth/blob/v1.12.20/eth/downloader/downloader.go) | | Symbol | `Downloader.processHeaders` | | Ref | `v1.12.20` |--- a/core-geth/eth/downloader/downloader.go
+++ b/etc/eth/downloader/downloader_pow.go
@@ -1,42 +1,17 @@
-// processHeaders takes batches of retrieved headers from an input channel and
+// processHeadersPoW takes batches of retrieved headers from an input channel and
// keeps processing and scheduling them into the header chain and downloader's
// queue until the stream ends or a failure occurs.
-func (d *Downloader) processHeaders(origin uint64, td, ttd *big.Int, beaconMode bool) error {
- // Keep a count of uncertain headers to roll back
+func (d *Downloader) processHeadersPoW(origin uint64, td *big.Int) error {
var (
- rollback uint64 // Zero means no rollback (fine as you can't unroll the genesis)
- rollbackErr error
mode = d.getMode()
+ gotHeaders = false // Wait for batches of headers to process
+ timer = time.NewTimer(time.Second)
)
- defer func() {
- if rollback > 0 {
- lastHeader, lastFastBlock, lastBlock := d.lightchain.CurrentHeader().Number, common.Big0, common.Big0
- if mode != LightSync {
- lastFastBlock = d.blockchain.CurrentSnapBlock().Number
- lastBlock = d.blockchain.CurrentBlock().Number
- }
- if err := d.lightchain.SetHead(rollback - 1); err != nil { // -1 to target the parent of the first uncertain block
- // We're already unwinding the stack, only print the error to make it more visible
- log.Error("Failed to roll back chain segment", "head", rollback-1, "err", err)
- }
- curFastBlock, curBlock := common.Big0, common.Big0
- if mode != LightSync {
- curFastBlock = d.blockchain.CurrentSnapBlock().Number
- curBlock = d.blockchain.CurrentBlock().Number
- }
- log.Warn("Rolled back chain segment",
- "header", fmt.Sprintf("%d->%d", lastHeader, d.lightchain.CurrentHeader().Number),
- "snap", fmt.Sprintf("%d->%d", lastFastBlock, curFastBlock),
- "block", fmt.Sprintf("%d->%d", lastBlock, curBlock), "reason", rollbackErr)
- }
- }()
- // Wait for batches of headers to process
- gotHeaders := false
+ defer timer.Stop()
for {
select {
case <-d.cancelCh:
- rollbackErr = errCanceled
return errCanceled
case task := <-d.headerProcCh:
@@ -49,14 +24,6 @@
case <-d.cancelCh:
}
}
- select {
- case d.totalDiffCh <- struct{}{}:
- case <-d.cancelCh:
- }
- // If we're in legacy sync mode, we need to check total difficulty
- // violations from malicious peers. That is not needed in beacon
- // mode and we can skip to terminating sync.
- if !beaconMode {
// If no headers were retrieved at all, the peer violated its TD promise that it had a
// better chain compared to ours. The only exception is if its promised blocks were
// already imported by other means (e.g. fetcher):
@@ -69,12 +36,10 @@
// L: Sync begins, and finds common ancestor at 11
// L: Request new headers up from 11 (R's TD was higher, it must have something)
// R: Nothing to give
- if mode != LightSync {
head := d.blockchain.CurrentBlock()
- if !gotHeaders && d.td.Cmp(d.blockchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
+ if !gotHeaders && td.Cmp(d.blockchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
return errStallingPeer
}
- }
// If snap or light syncing, ensure promised headers are indeed delivered. This is
// needed to detect scenarios where an attacker feeds a bad pivot and then bails out
// of delivering the post-pivot blocks that would flag the invalid content.
@@ -82,26 +47,22 @@
// This check cannot be executed "as is" for full imports, since blocks may still be
// queued for processing when the header download completes. However, as long as the
// peer gave us something useful, we're already happy/progressed (above check).
- if mode == SnapSync || mode == LightSync {
- head := d.lightchain.CurrentHeader()
- if d.td.Cmp(d.lightchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
+ if mode == ethconfig.SnapSync {
+ head := d.blockchain.CurrentHeader()
+ if td.Cmp(d.blockchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
return errStallingPeer
}
}
- }
- // Disable any rollback and return
- rollback = 0
return nil
}
// Otherwise split the chunk of headers into batches and process them
- headers, hashes := task.headers, task.hashes
+ headers, hashes, scheduled := task.headers, task.hashes, false
gotHeaders = true
for len(headers) > 0 {
// Terminate if something failed in between processing chunks
select {
case <-d.cancelCh:
- rollbackErr = errCanceled
return errCanceled
default:
}
@@ -113,96 +74,53 @@
chunkHeaders := headers[:limit]
chunkHashes := hashes[:limit]
- // In case of header only syncing, validate the chunk immediately
- if mode == SnapSync || mode == LightSync {
- // If we're importing pure headers, verify based on their recentness
- var pivot uint64
-
- d.pivotLock.RLock()
- if d.pivotHeader != nil {
- pivot = d.pivotHeader.Number.Uint64()
- }
- d.pivotLock.RUnlock()
-
- frequency := fsHeaderCheckFrequency
- if chunkHeaders[len(chunkHeaders)-1].Number.Uint64()+uint64(fsHeaderForceVerify) > pivot {
- frequency = 1
- }
- // Although the received headers might be all valid, a legacy
- // PoW/PoA sync must not accept post-merge headers. Make sure
- // that any transition is rejected at this point.
- var (
- rejected []*types.Header
- td *big.Int
- )
- if !beaconMode && ttd != nil {
- td = d.blockchain.GetTd(chunkHeaders[0].ParentHash, chunkHeaders[0].Number.Uint64()-1)
- if td == nil {
- // This should never really happen, but handle gracefully for now
- log.Error("Failed to retrieve parent header TD", "number", chunkHeaders[0].Number.Uint64()-1, "hash", chunkHeaders[0].ParentHash)
- return fmt.Errorf("%w: parent TD missing", errInvalidChain)
- }
- for i, header := range chunkHeaders {
- td = new(big.Int).Add(td, header.Difficulty)
- if td.Cmp(ttd) >= 0 {
- // Terminal total difficulty reached, allow the last header in
- if new(big.Int).Sub(td, header.Difficulty).Cmp(ttd) < 0 {
- chunkHeaders, rejected = chunkHeaders[:i+1], chunkHeaders[i+1:]
- if len(rejected) > 0 {
- // Make a nicer user log as to the first TD truly rejected
- td = new(big.Int).Add(td, rejected[0].Difficulty)
- }
- } else {
- chunkHeaders, rejected = chunkHeaders[:i], chunkHeaders[i:]
- }
- break
- }
- }
- }
- if len(chunkHeaders) > 0 {
- if n, err := d.lightchain.InsertHeaderChain(chunkHeaders, frequency); err != nil {
- rollbackErr = err
-
- // If some headers were inserted, track them as uncertain
- if mode == SnapSync && n > 0 && rollback == 0 {
- rollback = chunkHeaders[0].Number.Uint64()
- }
- log.Warn("Invalid header encountered", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err)
+ // Split the headers around the chain cutoff
+ var cutoff int
+ if mode == ethconfig.SnapSync && d.chainCutoffNumber != 0 {
+ cutoff = sort.Search(len(chunkHeaders), func(i int) bool {
+ return chunkHeaders[i].Number.Uint64() >= d.chainCutoffNumber
+ })
+ }
+ // Insert the header chain into the ancient store (with block bodies and
+ // receipts set to nil) if they fall before the cutoff.
+ if mode == ethconfig.SnapSync && cutoff != 0 {
+ if n, err := d.blockchain.InsertHeadersBeforeCutoff(chunkHeaders[:cutoff]); err != nil {
+ log.Warn("Failed to insert ancient header chain", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err)
return fmt.Errorf("%w: %v", errInvalidChain, err)
}
- // All verifications passed, track all headers within the allowed limits
- if mode == SnapSync {
- head := chunkHeaders[len(chunkHeaders)-1].Number.Uint64()
- if head-rollback > uint64(fsHeaderSafetyNet) {
- rollback = head - uint64(fsHeaderSafetyNet)
- } else {
- rollback = 1
- }
- }
+ log.Debug("Inserted headers before cutoff", "number", chunkHeaders[cutoff-1].Number, "hash", chunkHashes[cutoff-1])
}
- if len(rejected) != 0 {
- // Merge threshold reached, stop importing, but don't roll back
- rollback = 0
-
- log.Info("Legacy sync reached merge threshold", "number", rejected[0].Number, "hash", rejected[0].Hash(), "td", td, "ttd", ttd)
- return ErrMergeTransition
+ // In snap sync, insert headers into the header chain so that TDs
+ // are computed and persisted. Without this, headers exist in the
+ // database but have no associated TD, breaking PoW sync operations
+ // that rely on TD comparisons (cf. core-geth processHeaders).
+ if mode == ethconfig.SnapSync {
+ if n, err := d.blockchain.InsertHeaderChain(chunkHeaders); err != nil {
+ log.Warn("Failed to insert header chain", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err)
+ return fmt.Errorf("%w: %v", errInvalidChain, err)
}
}
- // Unless we're doing light chains, schedule the headers for associated content retrieval
- if mode == FullSync || mode == SnapSync {
// If we've reached the allowed number of pending headers, stall a bit
for d.queue.PendingBodies() >= maxQueuedHeaders || d.queue.PendingReceipts() >= maxQueuedHeaders {
+ timer.Reset(time.Second)
select {
case <-d.cancelCh:
- rollbackErr = errCanceled
return errCanceled
- case <-time.After(time.Second):
+ case <-timer.C:
+ }
}
+ // Otherwise, schedule the headers for content retrieval (block bodies and
+ // potentially receipts in snap sync).
+ //
+ // Skip the bodies/receipts retrieval scheduling before the cutoff in snap
+ // sync if chain pruning is configured.
+ if mode == ethconfig.SnapSync && cutoff != 0 {
+ chunkHeaders = chunkHeaders[cutoff:]
+ chunkHashes = chunkHashes[cutoff:]
}
- // Otherwise insert the headers for content retrieval
- inserts := d.queue.Schedule(chunkHeaders, chunkHashes, origin)
- if len(inserts) != len(chunkHeaders) {
- rollbackErr = fmt.Errorf("stale headers: len inserts %v len(chunk) %v", len(inserts), len(chunkHeaders))
+ if len(chunkHeaders) > 0 {
+ scheduled = true
+ if d.queue.Schedule(chunkHeaders, chunkHashes, origin+uint64(cutoff)) != len(chunkHeaders) {
return fmt.Errorf("%w: stale headers", errBadPeer)
}
}
@@ -218,6 +136,7 @@
d.syncStatsLock.Unlock()
// Signal the content downloaders of the availability of new tasks
+ if scheduled {
for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} {
select {
case ch <- true:
@@ -225,5 +144,6 @@
}
}
}
+ }
}
}