Skip to content

Downloader.syncWithPeer

Source (upstream pre-purge) Current
File eth/downloader/downloader.go downloader_pow.go
Symbol Downloader.syncWithPeer Downloader.syncWithPeer
Ref 45baf2111~1 etc/v1.17.3-full-node

Based on the !beaconMode branch of pre-purge syncWithPeer. Removed skeleton sync, TTD checks, and beacon-related logic. Added upstream's chainCutoffNumber / InsertHeadersBeforeCutoff path for ancient headers. Core-geth ref: same function name, same peer-based flow.

3-way merge — purge → getc ← upstream

pre-purgecore-geth≈ adapted (origin inferred by similarity)fork-only
↗// syncWithPeer starts a block synchronization based on the hash chain from the
// specified peer and head hash. This is for PoW networks only (no beacon mode).
func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.Int) (err error) {
≈ mode := d.getMode()
d.feed.Send(SyncEvent{Type: SyncStarted, Mode: mode})
↗ defer func() {
// reset on error
if err != nil {
d.feed.Send(SyncEvent{Type: SyncFailed, Mode: mode, Err: err})
↗ } else {
≈ latest := d.blockchain.CurrentHeader()
d.feed.Send(SyncEvent{Type: SyncCompleted, Mode: mode, Latest: latest})
↗ }
}()
≈ log.Debug("Synchronising with the network", "peer", p.id, "eth", p.version, "head", hash, "td", td, "mode", mode)
↗ defer func(start time.Time) {
log.Debug("Synchronisation terminated", "elapsed", common.PrettyDuration(time.Since(start)))
}(time.Now())
↗ // Look up the sync boundaries: the common ancestor and the target block
// In legacy mode, use the master peer to retrieve the headers from
≈ latest, pivot, err := d.fetchHead(p)
≈ if err != nil {
≈ return err
↗ }
// If no pivot block was returned, the head is below the min full block
// threshold (i.e. new chain). In that case we won't really snap sync
// anyway, but still need a valid pivot block to avoid some code hitting
// nil panics on access.
≈ if mode == ethconfig.SnapSync && pivot == nil {
↗ pivot = d.blockchain.CurrentBlock()
}
height := latest.Number.Uint64()
// In legacy mode, reach out to the network and find the ancestor
≈ origin, err := d.findAncestor(p, latest)
≈ if err != nil {
≈ return err
↗ }
d.syncStatsLock.Lock()
if d.syncStatsChainHeight <= origin || d.syncStatsChainOrigin > origin {
d.syncStatsChainOrigin = origin
}
d.syncStatsChainHeight = height
d.syncStatsLock.Unlock()
↗ // Ensure our origin point is below any snap sync pivot point
≈ if mode == ethconfig.SnapSync {
↗ if height <= uint64(fsMinFullBlocks) {
origin = 0
} else {
pivotNumber := pivot.Number.Uint64()
if pivotNumber <= origin {
origin = pivotNumber - 1
}
// Write out the pivot into the database so a rollback beyond it will
// reenable snap sync
rawdb.WriteLastPivotNumber(d.stateDB, pivotNumber)
}
}
d.committed.Store(true)
≈ if mode == ethconfig.SnapSync && pivot.Number.Uint64() != 0 {
↗ d.committed.Store(false)
}
≈ if mode == ethconfig.SnapSync {
// Set the ancient data limitation. Legacy sync, use the best
// announcement we have from the remote peer.
≈ if height > fullMaxForkAncestry+1 {
≈ d.ancientLimit = height - fullMaxForkAncestry - 1
↗ } else {
≈ d.ancientLimit = 0
↗ }
// Extend the ancient chain segment range if the ancient limit is even
// below the pre-configured chain cutoff.
if d.chainCutoffNumber != 0 && d.chainCutoffNumber > d.ancientLimit {
≈ d.ancientLimit = d.chainCutoffNumber
log.Info("Extend the ancient range with configured cutoff", "cutoff", d.chainCutoffNumber)
}
↗ frozen, _ := d.stateDB.Ancients() // Ignore the error here since light client can also hit here.
↗ // If a part of blockchain data has already been written into active store,
// disable the ancient style insertion explicitly.
≈ if origin >= frozen && origin != 0 {
↗ d.ancientLimit = 0
var ancient string
≈ if frozen == 0 {
ancient = "null"
≈ } else {
ancient = fmt.Sprintf("%d", frozen-1)
}
≈ log.Info("Disabling direct-ancient mode", "origin", origin, "ancient", ancient)
↗ } else if d.ancientLimit > 0 {
log.Debug("Enabling direct-ancient mode", "ancient", d.ancientLimit)
}
// Rewind the ancient store and blockchain if reorg happens.
if origin+1 < frozen {
≈ if err := d.blockchain.SetHead(origin); err != nil {
↗ return err
}
log.Info("Truncated excess ancient chain segment", "oldhead", frozen-1, "newhead", origin)
}
}
// Skip ancient chain segments if Geth is running with a configured chain cutoff.
// These segments are not guaranteed to be available in the network.
≈ chainOffset := origin + 1
if mode == ethconfig.SnapSync && d.chainCutoffNumber != 0 {
if chainOffset < d.chainCutoffNumber {
chainOffset = d.chainCutoffNumber
log.Info("Skip chain segment before cutoff", "origin", origin, "cutoff", d.chainCutoffNumber)
}
↗ }
// Initiate the sync using a concurrent header and content retrieval algorithm
≈ d.queue.Prepare(chainOffset, mode)
// Create queueWithHeaders for header fetching
qwh := newQueueWithHeaders(d.queue)
qwh.ResetHeaders()
// In legacy mode, headers are retrieved from the network
↗ fetchers := []func() error{
≈ func() error { return d.fetchHeadersPoW(p, origin+1, latest.Number.Uint64(), qwh) },
≈ func() error { return d.fetchBodies(chainOffset) }, // Bodies are retrieved during normal and snap sync
func() error { return d.fetchReceipts(chainOffset) }, // Receipts are retrieved during snap sync
≈ func() error { return d.processHeadersPoW(origin+1, td) },
↗ }
≈ if mode == ethconfig.SnapSync {
↗ d.pivotLock.Lock()
d.pivotHeader = pivot
d.pivotLock.Unlock()
↗ fetchers = append(fetchers, func() error { return d.processSnapSyncContent() })
≈ } else if mode == ethconfig.FullSync {
≈ fetchers = append(fetchers, func() error { return d.processFullSyncContent() })
↗ }
return d.spawnSync(fetchers)
}
core-geth validation — +52 -113 | | | |---|---| | File | [`downloader.go`](https://github.com/etclabscore/core-geth/blob/v1.12.20/eth/downloader/downloader.go) | | Symbol | `Downloader.syncWithPeer` | | Ref | `v1.12.20` | 
--- a/core-geth/eth/downloader/downloader.go
+++ b/etc/eth/downloader/downloader_pow.go
@@ -1,89 +1,43 @@
 // syncWithPeer starts a block synchronization based on the hash chain from the
-// specified peer and head hash.
-func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td, ttd *big.Int, beaconMode bool) (err error) {
-   d.mux.Post(StartEvent{})
+// specified peer and head hash. This is for PoW networks only (no beacon mode).
+func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.Int) (err error) {
+   mode := d.getMode()
+   d.feed.Send(SyncEvent{Type: SyncStarted, Mode: mode})
    defer func() {
        // reset on error
        if err != nil {
-           d.mux.Post(FailedEvent{err})
+           d.feed.Send(SyncEvent{Type: SyncFailed, Mode: mode, Err: err})
        } else {
-           latest := d.lightchain.CurrentHeader()
-           d.mux.Post(DoneEvent{latest})
+           latest := d.blockchain.CurrentHeader()
+           d.feed.Send(SyncEvent{Type: SyncCompleted, Mode: mode, Latest: latest})
        }
    }()
-   mode := d.getMode()

-   if !beaconMode {
        log.Debug("Synchronising with the network", "peer", p.id, "eth", p.version, "head", hash, "td", td, "mode", mode)
-   } else {
-       log.Debug("Backfilling with the network", "mode", mode)
-   }
    defer func(start time.Time) {
        log.Debug("Synchronisation terminated", "elapsed", common.PrettyDuration(time.Since(start)))
    }(time.Now())

    // Look up the sync boundaries: the common ancestor and the target block
-   var latest, pivot, final *types.Header
-   if !beaconMode {
        // In legacy mode, use the master peer to retrieve the headers from
-       latest, pivot, err = d.fetchHead(p)
+   latest, pivot, err := d.fetchHead(p)
        if err != nil {
            return err
        }
-   } else {
-       // In beacon mode, use the skeleton chain to retrieve the headers from
-       latest, _, final, err = d.skeleton.Bounds()
-       if err != nil {
-           return err
-       }
-       if latest.Number.Uint64() > uint64(fsMinFullBlocks) {
-           number := latest.Number.Uint64() - uint64(fsMinFullBlocks)
-
-           // Retrieve the pivot header from the skeleton chain segment but
-           // fallback to local chain if it's not found in skeleton space.
-           if pivot = d.skeleton.Header(number); pivot == nil {
-               _, oldest, _, _ := d.skeleton.Bounds() // error is already checked
-               if number < oldest.Number.Uint64() {
-                   count := int(oldest.Number.Uint64() - number) // it's capped by fsMinFullBlocks
-                   headers := d.readHeaderRange(oldest, count)
-                   if len(headers) == count {
-                       pivot = headers[len(headers)-1]
-                       log.Warn("Retrieved pivot header from local", "number", pivot.Number, "hash", pivot.Hash(), "latest", latest.Number, "oldest", oldest.Number)
-                   }
-               }
-           }
-           // Print an error log and return directly in case the pivot header
-           // is still not found. It means the skeleton chain is not linked
-           // correctly with local chain.
-           if pivot == nil {
-               log.Error("Pivot header is not found", "number", number)
-               return errNoPivotHeader
-           }
-       }
-   }
    // If no pivot block was returned, the head is below the min full block
    // threshold (i.e. new chain). In that case we won't really snap sync
    // anyway, but still need a valid pivot block to avoid some code hitting
    // nil panics on access.
-   if mode == SnapSync && pivot == nil {
+   if mode == ethconfig.SnapSync && pivot == nil {
        pivot = d.blockchain.CurrentBlock()
    }
    height := latest.Number.Uint64()

-   var origin uint64
-   if !beaconMode {
        // In legacy mode, reach out to the network and find the ancestor
-       origin, err = d.findAncestor(p, latest)
+   origin, err := d.findAncestor(p, latest)
        if err != nil {
            return err
        }
-   } else {
-       // In beacon mode, use the skeleton chain for the ancestor lookup
-       origin, err = d.findBeaconAncestor()
-       if err != nil {
-           return err
-       }
-   }
    d.syncStatsLock.Lock()
    if d.syncStatsChainHeight <= origin || d.syncStatsChainOrigin > origin {
        d.syncStatsChainOrigin = origin
@@ -92,7 +46,7 @@
    d.syncStatsLock.Unlock()

    // Ensure our origin point is below any snap sync pivot point
-   if mode == SnapSync {
+   if mode == ethconfig.SnapSync {
        if height <= uint64(fsMinFullBlocks) {
            origin = 0
        } else {
@@ -106,93 +60,78 @@
        }
    }
    d.committed.Store(true)
-   if mode == SnapSync && pivot.Number.Uint64() != 0 {
+   if mode == ethconfig.SnapSync && pivot.Number.Uint64() != 0 {
        d.committed.Store(false)
    }
-   if mode == SnapSync {
-       // Set the ancient data limitation. If we are running snap sync, all block
-       // data older than ancientLimit will be written to the ancient store. More
-       // recent data will be written to the active database and will wait for the
-       // freezer to migrate.
-       //
-       // If the network is post-merge, use either the last announced finalized
-       // block as the ancient limit, or if we haven't yet received one, the head-
-       // a max fork ancestry limit. One quirky case if we've already passed the
-       // finalized block, in which case the skeleton.Bounds will return nil and
-       // we'll revert to head - 90K. That's fine, we're finishing sync anyway.
-       //
-       // For non-merged networks, if there is a checkpoint available, then calculate
-       // the ancientLimit through that. Otherwise calculate the ancient limit through
-       // the advertised height of the remote peer. This most is mostly a fallback for
-       // legacy networks, but should eventually be dropped. TODO(karalabe).
-       if beaconMode {
-           // Beacon sync, use the latest finalized block as the ancient limit
-           // or a reasonable height if no finalized block is yet announced.
-           if final != nil {
-               d.ancientLimit = final.Number.Uint64()
-           } else if height > fullMaxForkAncestry+1 {
-               d.ancientLimit = height - fullMaxForkAncestry - 1
-           } else {
-               d.ancientLimit = 0
-           }
-       } else {
-           // Legacy sync, use any hardcoded checkpoints or the best announcement
-           // we have from the remote peer. TODO(karalabe): Drop this pathway.
-           if d.checkpoint != 0 && d.checkpoint > fullMaxForkAncestry+1 {
-               d.ancientLimit = d.checkpoint
-           } else if height > fullMaxForkAncestry+1 {
+   if mode == ethconfig.SnapSync {
+       // Set the ancient data limitation. Legacy sync, use the best
+       // announcement we have from the remote peer.
+       if height > fullMaxForkAncestry+1 {
                d.ancientLimit = height - fullMaxForkAncestry - 1
            } else {
                d.ancientLimit = 0
            }
+       // Extend the ancient chain segment range if the ancient limit is even
+       // below the pre-configured chain cutoff.
+       if d.chainCutoffNumber != 0 && d.chainCutoffNumber > d.ancientLimit {
+           d.ancientLimit = d.chainCutoffNumber
+           log.Info("Extend the ancient range with configured cutoff", "cutoff", d.chainCutoffNumber)
        }
        frozen, _ := d.stateDB.Ancients() // Ignore the error here since light client can also hit here.

        // If a part of blockchain data has already been written into active store,
        // disable the ancient style insertion explicitly.
-       if origin >= frozen && frozen != 0 {
+       if origin >= frozen && origin != 0 {
            d.ancientLimit = 0
-           log.Info("Disabling direct-ancient mode", "origin", origin, "ancient", frozen-1)
+           var ancient string
+           if frozen == 0 {
+               ancient = "null"
+           } else {
+               ancient = fmt.Sprintf("%d", frozen-1)
+           }
+           log.Info("Disabling direct-ancient mode", "origin", origin, "ancient", ancient)
        } else if d.ancientLimit > 0 {
            log.Debug("Enabling direct-ancient mode", "ancient", d.ancientLimit)
        }
        // Rewind the ancient store and blockchain if reorg happens.
        if origin+1 < frozen {
-           if err := d.lightchain.SetHead(origin); err != nil {
+           if err := d.blockchain.SetHead(origin); err != nil {
                return err
            }
            log.Info("Truncated excess ancient chain segment", "oldhead", frozen-1, "newhead", origin)
        }
    }
-   // Initiate the sync using a concurrent header and content retrieval algorithm
-   d.queue.Prepare(origin+1, mode)
-   if d.syncInitHook != nil {
-       d.syncInitHook(origin, height)
+   // Skip ancient chain segments if Geth is running with a configured chain cutoff.
+   // These segments are not guaranteed to be available in the network.
+   chainOffset := origin + 1
+   if mode == ethconfig.SnapSync && d.chainCutoffNumber != 0 {
+       if chainOffset < d.chainCutoffNumber {
+           chainOffset = d.chainCutoffNumber
+           log.Info("Skip chain segment before cutoff", "origin", origin, "cutoff", d.chainCutoffNumber)
    }
-   var headerFetcher func() error
-   if !beaconMode {
-       // In legacy mode, headers are retrieved from the network
-       headerFetcher = func() error { return d.fetchHeaders(p, origin+1, latest.Number.Uint64()) }
-   } else {
-       // In beacon mode, headers are served by the skeleton syncer
-       headerFetcher = func() error { return d.fetchBeaconHeaders(origin + 1) }
    }
+   // Initiate the sync using a concurrent header and content retrieval algorithm
+   d.queue.Prepare(chainOffset, mode)
+
+   // Create queueWithHeaders for header fetching
+   qwh := newQueueWithHeaders(d.queue)
+   qwh.ResetHeaders()
+
+   // In legacy mode, headers are retrieved from the network
    fetchers := []func() error{
-       headerFetcher, // Headers are always retrieved
-       func() error { return d.fetchBodies(origin+1, beaconMode) },   // Bodies are retrieved during normal and snap sync
-       func() error { return d.fetchReceipts(origin+1, beaconMode) }, // Receipts are retrieved during snap sync
-       func() error { return d.processHeaders(origin+1, td, ttd, beaconMode) },
+       func() error { return d.fetchHeadersPoW(p, origin+1, latest.Number.Uint64(), qwh) },
+       func() error { return d.fetchBodies(chainOffset) },   // Bodies are retrieved during normal and snap sync
+       func() error { return d.fetchReceipts(chainOffset) }, // Receipts are retrieved during snap sync
+       func() error { return d.processHeadersPoW(origin+1, td) },
    }
-   if mode == SnapSync {
+   if mode == ethconfig.SnapSync {
        d.pivotLock.Lock()
        d.pivotHeader = pivot
        d.pivotLock.Unlock()

        fetchers = append(fetchers, func() error { return d.processSnapSyncContent() })
-   } else if mode == FullSync {
-       fetchers = append(fetchers, func() error { return d.processFullSyncContent(ttd, beaconMode) })
+   } else if mode == ethconfig.FullSync {
+       fetchers = append(fetchers, func() error { return d.processFullSyncContent() })
    }
-   fetchers = append(fetchers, func() error { return d.fetchTotalDifficulty(p, latest) })
-
    return d.spawnSync(fetchers)
 }

← Sync & Downloader