eth/handler.go — Version Comparison¶
| v1.16.8 | v1.17.3 | |
|---|---|---|
| Branch | etc/v1.16.8-full-node |
etc/v1.17.3-full-node |
| Delta | +98 -33 | +101 -31 |
ETC delta on v1.16.8 (+98 -33)
diff --git a/eth/handler.go b/eth/handler.go
index ff970e2ba..2130909ae 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -30,6 +30,7 @@ import (
"github.com/dchest/siphash"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
+ "github.com/ethereum/go-ethereum/core/forkid"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
@@ -97,23 +98,31 @@ type txPool interface {
// handlerConfig is the collection of initialization parameters to create a full
// node network handler.
type handlerConfig struct {
- NodeID enode.ID // P2P node ID used for tx propagation topology
- Database ethdb.Database // Database for direct sync insertions
- Chain *core.BlockChain // Blockchain to serve data from
- TxPool txPool // Transaction pool to propagate from
- Network uint64 // Network identifier to advertise
- Sync ethconfig.SyncMode // Whether to snap or full sync
- BloomCache uint64 // Megabytes to alloc for snap sync bloom
- EventMux *event.TypeMux // Legacy event mux, deprecate for `feed`
- RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
+ NodeID enode.ID // P2P node ID used for tx propagation topology
+ Database ethdb.Database // Database for direct sync insertions
+ Chain *core.BlockChain // Blockchain to serve data from
+ TxPool txPool // Transaction pool to propagate from
+ Network uint64 // Network identifier to advertise
+ Sync ethconfig.SyncMode // Whether to snap or full sync
+ BloomCache uint64 // Megabytes to alloc for snap sync bloom
+ EventMux *event.TypeMux // Legacy event mux, deprecate for `feed`
+ RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
+ IsPow bool // Whether this is a perpetual PoW chain (uses TD-based sync)
+ MESSForceEnable bool // Force enable MESS (ECBP-1100) artificial finality
+ MESSForceDisable bool // Force disable MESS (ECBP-1100) artificial finality
}
type handler struct {
- nodeID enode.ID
- networkID uint64
+ nodeID enode.ID
+ networkID uint64
+ forkFilter forkid.Filter // Fork ID filter, constant across the lifetime of the node
snapSync atomic.Bool // Flag whether snap sync is enabled (gets disabled if we already have blocks)
synced atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
+ isPow bool // Whether this is a perpetual PoW chain (uses TD-based sync)
+
+ messForceEnable bool // Force enable MESS (emergency override)
+ messForceDisable bool // Force disable MESS (never enable)
database ethdb.Database
txpool txPool
@@ -122,13 +131,16 @@ type handler struct {
downloader *downloader.Downloader
txFetcher *fetcher.TxFetcher
+ blockFetcher *fetcher.BlockFetcher // Block fetcher for PoW chains
peers *peerSet
txBroadcastKey [16]byte
+ chainSync *chainSyncer // PoW chain syncer (nil for PoS chains)
- eventMux *event.TypeMux
- txsCh chan core.NewTxsEvent
- txsSub event.Subscription
- blockRange *blockRangeState
+ eventMux *event.TypeMux
+ txsCh chan core.NewTxsEvent
+ txsSub event.Subscription
+ minedBlockSub *event.TypeMuxSubscription // PoW mined block subscription (nil for PoS)
+ blockRange *blockRangeState
requiredBlocks map[uint64]common.Hash
@@ -148,18 +160,22 @@ func newHandler(config *handlerConfig) (*handler, error) {
config.EventMux = new(event.TypeMux) // Nicety initialization for tests
}
h := &handler{
- nodeID: config.NodeID,
- networkID: config.Network,
- eventMux: config.EventMux,
- database: config.Database,
- txpool: config.TxPool,
- chain: config.Chain,
- peers: newPeerSet(),
- txBroadcastKey: newBroadcastChoiceKey(),
- requiredBlocks: config.RequiredBlocks,
- quitSync: make(chan struct{}),
- handlerDoneCh: make(chan struct{}),
- handlerStartCh: make(chan struct{}),
+ nodeID: config.NodeID,
+ networkID: config.Network,
+ forkFilter: forkid.NewFilter(config.Chain),
+ eventMux: config.EventMux,
+ database: config.Database,
+ txpool: config.TxPool,
+ chain: config.Chain,
+ peers: newPeerSet(),
+ txBroadcastKey: newBroadcastChoiceKey(),
+ requiredBlocks: config.RequiredBlocks,
+ quitSync: make(chan struct{}),
+ handlerDoneCh: make(chan struct{}),
+ handlerStartCh: make(chan struct{}),
+ isPow: config.IsPow,
+ messForceEnable: config.MESSForceEnable,
+ messForceDisable: config.MESSForceDisable,
}
if config.Sync == ethconfig.FullSync {
// The database seems empty as the current block is the genesis. Yet the snap
@@ -262,9 +278,24 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
}
// Execute the Ethereum handshake
- if err := peer.Handshake(h.networkID, h.chain, h.blockRange.currentRange()); err != nil {
- peer.Log().Debug("Ethereum handshake failed", "err", err)
- return err
+ if h.isPow {
+ var (
+ genesis = h.chain.Genesis()
+ head = h.chain.CurrentHeader()
+ hash = head.Hash()
+ number = head.Number.Uint64()
+ td = h.chain.GetTd(hash, number)
+ )
+ forkID := forkid.NewID(h.chain.Config(), genesis, number, head.Time)
+ if err := peer.Handshake68PoW(h.networkID, td, hash, genesis.Hash(), forkID, h.forkFilter); err != nil {
+ peer.Log().Debug("Ethereum handshake failed", "err", err)
+ return err
+ }
+ } else {
+ if err := peer.Handshake(h.networkID, h.chain, h.blockRange.currentRange()); err != nil {
+ peer.Log().Debug("Ethereum handshake failed", "err", err)
+ return err
+ }
}
reject := false // reserved peer slots
if h.snapSync.Load() {
@@ -307,6 +338,10 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
return err
}
}
+ // Notify ETC chain syncer about the new peer
+ if h.isPow {
+ h.chainSync.handlePeerEvent()
+ }
// Propagate existing transactions. new transactions appearing
// after this will be sent via broadcasts.
h.syncTransactions(peer)
@@ -441,7 +476,26 @@ func (h *handler) Start(maxPeers int) {
go h.blockRangeLoop(h.blockRange)
// start sync handlers
- h.txFetcher.Start()
+ // For PoW chains, txFetcher lifecycle is managed by chainSyncer.loop()
+ if !h.isPow {
+ h.txFetcher.Start()
+ }
+
+ // Start PoW chain syncer and block fetcher if this is a perpetual PoW chain
+ if h.isPow {
+ h.startBlockFetcher()
+ h.chainSync = newChainSyncer(h)
+ h.wg.Add(1)
+ go h.chainSync.loop()
+ // Start mined block broadcast loop
+ h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{})
+ h.wg.Add(1)
+ go h.minedBroadcastLoop()
+ // Start MESS safety loop for stale head detection
+ h.wg.Add(1)
+ go h.messSafetyLoop()
+ log.Info("Starting PoW chain syncer and block fetcher")
+ }
// start peer handler tracker
h.wg.Add(1)
@@ -450,9 +504,20 @@ func (h *handler) Start(maxPeers int) {
func (h *handler) Stop() {
h.txsSub.Unsubscribe() // quits txBroadcastLoop
+ if h.minedBlockSub != nil {
+ h.minedBlockSub.Unsubscribe() // quits minedBroadcastLoop
+ }
h.blockRange.stop()
- h.txFetcher.Stop()
- h.downloader.Terminate()
+
+ // For PoW chains, fetcher/downloader lifecycle is managed by chainSyncer.loop() defers.
+ // The shutdown flow is: close(quitSync) → loop() exits → defers run Stop/Terminate → wg.Wait() returns.
+ if !h.isPow {
+ h.txFetcher.Stop()
+ if h.blockFetcher != nil {
+ h.blockFetcher.Stop()
+ }
+ h.downloader.Terminate()
+ }
// Quit chainSync and txsync64.
// After this is done, no new peers will be accepted.
ETC delta on v1.17.3 (+101 -31)
diff --git a/eth/handler.go b/eth/handler.go
index 76df635fb..f6a8e037e 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -30,6 +30,7 @@ import (
"github.com/dchest/siphash"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
+ "github.com/ethereum/go-ethereum/core/forkid"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
@@ -100,20 +101,29 @@ type txPool interface {
// handlerConfig is the collection of initialization parameters to create a full
// node network handler.
type handlerConfig struct {
- NodeID enode.ID // P2P node ID used for tx propagation topology
- Database ethdb.Database // Database for direct sync insertions
- Chain *core.BlockChain // Blockchain to serve data from
- TxPool txPool // Transaction pool to propagate from
- Network uint64 // Network identifier to advertise
- Sync ethconfig.SyncMode // Whether to snap or full sync
- BloomCache uint64 // Megabytes to alloc for snap sync bloom
- RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
+ NodeID enode.ID // P2P node ID used for tx propagation topology
+ Database ethdb.Database // Database for direct sync insertions
+ Chain *core.BlockChain // Blockchain to serve data from
+ TxPool txPool // Transaction pool to propagate from
+ Network uint64 // Network identifier to advertise
+ Sync ethconfig.SyncMode // Whether to snap or full sync
+ BloomCache uint64 // Megabytes to alloc for snap sync bloom
+ RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
+ IsPow bool // Whether this is a perpetual PoW chain (uses TD-based sync)
+ MESSForceEnable bool // Force enable MESS (ECBP-1100) artificial finality
+ MESSForceDisable bool // Force disable MESS (ECBP-1100) artificial finality
}
type handler struct {
- nodeID enode.ID
- networkID uint64
- synced atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
+ nodeID enode.ID
+ networkID uint64
+ forkFilter forkid.Filter // Fork ID filter, constant across the lifetime of the node
+
+ synced atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
+ isPow bool // Whether this is a perpetual PoW chain (uses TD-based sync)
+
+ messForceEnable bool // Force enable MESS (emergency override)
+ messForceDisable bool // Force disable MESS (never enable)
database ethdb.Database
txpool txPool
@@ -122,12 +132,16 @@ type handler struct {
downloader *downloader.Downloader
txFetcher *fetcher.TxFetcher
+ blockFetcher *fetcher.BlockFetcher // Block fetcher for PoW chains
peers *peerSet
txBroadcastKey [16]byte
+ chainSync *chainSyncer // PoW chain syncer (nil for PoS chains)
- txsCh chan core.NewTxsEvent
- txsSub event.Subscription
- blockRange *blockRangeState
+ eventMux *event.TypeMux
+ txsCh chan core.NewTxsEvent
+ txsSub event.Subscription
+ minedBlockSub *event.TypeMuxSubscription // PoW mined block subscription (nil for PoS)
+ blockRange *blockRangeState
requiredBlocks map[uint64]common.Hash
@@ -143,17 +157,24 @@ type handler struct {
// newHandler returns a handler for all Ethereum chain management protocol.
func newHandler(config *handlerConfig) (*handler, error) {
h := &handler{
- nodeID: config.NodeID,
- networkID: config.Network,
- database: config.Database,
- txpool: config.TxPool,
- chain: config.Chain,
- peers: newPeerSet(),
- txBroadcastKey: newBroadcastChoiceKey(),
- requiredBlocks: config.RequiredBlocks,
- quitSync: make(chan struct{}),
- handlerDoneCh: make(chan struct{}),
- handlerStartCh: make(chan struct{}),
+ nodeID: config.NodeID,
+ networkID: config.Network,
+ forkFilter: forkid.NewFilter(config.Chain),
+ database: config.Database,
+ txpool: config.TxPool,
+ chain: config.Chain,
+ peers: newPeerSet(),
+ txBroadcastKey: newBroadcastChoiceKey(),
+ requiredBlocks: config.RequiredBlocks,
+ quitSync: make(chan struct{}),
+ handlerDoneCh: make(chan struct{}),
+ handlerStartCh: make(chan struct{}),
+ isPow: config.IsPow,
+ messForceEnable: config.MESSForceEnable,
+ messForceDisable: config.MESSForceDisable,
+ }
+ if config.IsPow {
+ h.eventMux = new(event.TypeMux)
}
// Construct the downloader (long sync)
h.downloader = downloader.New(config.Database, config.Sync, h.chain, h.removePeer, h.enableSyncedFeatures)
@@ -238,9 +259,24 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
}
// Execute the Ethereum handshake
- if err := peer.Handshake(h.networkID, h.chain, h.blockRange.currentRange()); err != nil {
- peer.Log().Debug("Ethereum handshake failed", "err", err)
- return err
+ if h.isPow {
+ var (
+ genesis = h.chain.Genesis()
+ head = h.chain.CurrentHeader()
+ hash = head.Hash()
+ number = head.Number.Uint64()
+ td = h.chain.GetTd(hash, number)
+ )
+ forkID := forkid.NewID(h.chain.Config(), genesis, number, head.Time)
+ if err := peer.Handshake68PoW(h.networkID, td, hash, genesis.Hash(), forkID, h.forkFilter); err != nil {
+ peer.Log().Debug("Ethereum handshake failed", "err", err)
+ return err
+ }
+ } else {
+ if err := peer.Handshake(h.networkID, h.chain, h.blockRange.currentRange()); err != nil {
+ peer.Log().Debug("Ethereum handshake failed", "err", err)
+ return err
+ }
}
reject := false // reserved peer slots
if h.downloader.ConfigSyncMode() == ethconfig.SnapSync {
@@ -283,6 +319,10 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
return err
}
}
+ // Notify ETC chain syncer about the new peer
+ if h.isPow {
+ h.chainSync.handlePeerEvent()
+ }
// Propagate existing transactions. new transactions appearing
// after this will be sent via broadcasts.
h.syncTransactions(peer)
@@ -417,7 +457,26 @@ func (h *handler) Start(maxPeers int) {
go h.blockRangeLoop(h.blockRange)
// start sync handlers
- h.txFetcher.Start()
+ // For PoW chains, txFetcher lifecycle is managed by chainSyncer.loop()
+ if !h.isPow {
+ h.txFetcher.Start()
+ }
+
+ // Start PoW chain syncer and block fetcher if this is a perpetual PoW chain
+ if h.isPow {
+ h.startBlockFetcher()
+ h.chainSync = newChainSyncer(h)
+ h.wg.Add(1)
+ go h.chainSync.loop()
+ // Start mined block broadcast loop
+ h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{})
+ h.wg.Add(1)
+ go h.minedBroadcastLoop()
+ // Start MESS safety loop for stale head detection
+ h.wg.Add(1)
+ go h.messSafetyLoop()
+ log.Info("Starting PoW chain syncer and block fetcher")
+ }
// start peer handler tracker
h.wg.Add(1)
@@ -426,9 +485,20 @@ func (h *handler) Start(maxPeers int) {
func (h *handler) Stop() {
h.txsSub.Unsubscribe() // quits txBroadcastLoop
+ if h.minedBlockSub != nil {
+ h.minedBlockSub.Unsubscribe() // quits minedBroadcastLoop
+ }
h.blockRange.stop()
- h.txFetcher.Stop()
- h.downloader.Terminate()
+
+ // For PoW chains, fetcher/downloader lifecycle is managed by chainSyncer.loop() defers.
+ // The shutdown flow is: close(quitSync) → loop() exits → defers run Stop/Terminate → wg.Wait() returns.
+ if !h.isPow {
+ h.txFetcher.Stop()
+ if h.blockFetcher != nil {
+ h.blockFetcher.Stop()
+ }
+ h.downloader.Terminate()
+ }
// Quit chainSync and txsync64.
// After this is done, no new peers will be accepted.