mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
consensus, eth: enforce always merged network
This commit is contained in:
Péter Szilágyi
parent
1655c83d24
commit
b443c3c215
15 changed files with 45 additions and 568 deletions
|
|
@ -1,110 +0,0 @@
|
|||
// Copyright 2021 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
package consensus
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/ethereum/go-ethereum/core/rawdb"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/rlp"
|
||||
)
|
||||
|
||||
// transitionStatus describes the status of eth1/2 transition. This switch
|
||||
// between modes is a one-way action which is triggered by corresponding
|
||||
// consensus-layer message.
|
||||
type transitionStatus struct {
|
||||
LeftPoW bool // The flag is set when the first NewHead message received
|
||||
EnteredPoS bool // The flag is set when the first FinalisedBlock message received
|
||||
}
|
||||
|
||||
// Merger is an internal help structure used to track the eth1/2 transition status.
|
||||
// It's a common structure can be used in both full node and light client.
|
||||
type Merger struct {
|
||||
db ethdb.KeyValueStore
|
||||
status transitionStatus
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewMerger creates a new Merger which stores its transition status in the provided db.
|
||||
func NewMerger(db ethdb.KeyValueStore) *Merger {
|
||||
var status transitionStatus
|
||||
blob := rawdb.ReadTransitionStatus(db)
|
||||
if len(blob) != 0 {
|
||||
if err := rlp.DecodeBytes(blob, &status); err != nil {
|
||||
log.Crit("Failed to decode the transition status", "err", err)
|
||||
}
|
||||
}
|
||||
return &Merger{
|
||||
db: db,
|
||||
status: status,
|
||||
}
|
||||
}
|
||||
|
||||
// ReachTTD is called whenever the first NewHead message received
|
||||
// from the consensus-layer.
|
||||
func (m *Merger) ReachTTD() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if m.status.LeftPoW {
|
||||
return
|
||||
}
|
||||
m.status = transitionStatus{LeftPoW: true}
|
||||
blob, err := rlp.EncodeToBytes(m.status)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("Failed to encode the transition status: %v", err))
|
||||
}
|
||||
rawdb.WriteTransitionStatus(m.db, blob)
|
||||
log.Info("Left PoW stage")
|
||||
}
|
||||
|
||||
// FinalizePoS is called whenever the first FinalisedBlock message received
|
||||
// from the consensus-layer.
|
||||
func (m *Merger) FinalizePoS() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if m.status.EnteredPoS {
|
||||
return
|
||||
}
|
||||
m.status = transitionStatus{LeftPoW: true, EnteredPoS: true}
|
||||
blob, err := rlp.EncodeToBytes(m.status)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("Failed to encode the transition status: %v", err))
|
||||
}
|
||||
rawdb.WriteTransitionStatus(m.db, blob)
|
||||
log.Info("Entered PoS stage")
|
||||
}
|
||||
|
||||
// TDDReached reports whether the chain has left the PoW stage.
|
||||
func (m *Merger) TDDReached() bool {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
return m.status.LeftPoW
|
||||
}
|
||||
|
||||
// PoSFinalized reports whether the chain has entered the PoS stage.
|
||||
func (m *Merger) PoSFinalized() bool {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
return m.status.EnteredPoS
|
||||
}
|
||||
|
|
@ -74,7 +74,6 @@ type Ethereum struct {
|
|||
handler *handler
|
||||
ethDialCandidates enode.Iterator
|
||||
snapDialCandidates enode.Iterator
|
||||
merger *consensus.Merger
|
||||
|
||||
// DB interfaces
|
||||
chainDb ethdb.Database // Block chain database
|
||||
|
|
@ -158,7 +157,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
|
|||
}
|
||||
eth := &Ethereum{
|
||||
config: config,
|
||||
merger: consensus.NewMerger(chainDb),
|
||||
chainDb: chainDb,
|
||||
eventMux: stack.EventMux(),
|
||||
accountManager: stack.AccountManager(),
|
||||
|
|
@ -240,7 +238,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
|
|||
Database: chainDb,
|
||||
Chain: eth.blockchain,
|
||||
TxPool: eth.txPool,
|
||||
Merger: eth.merger,
|
||||
Network: networkID,
|
||||
Sync: config.SyncMode,
|
||||
BloomCache: uint64(cacheLimit),
|
||||
|
|
@ -487,7 +484,6 @@ func (s *Ethereum) Synced() bool { return s.handler.synced
|
|||
func (s *Ethereum) SetSynced() { s.handler.enableSyncedFeatures() }
|
||||
func (s *Ethereum) ArchiveMode() bool { return s.config.NoPruning }
|
||||
func (s *Ethereum) BloomIndexer() *core.ChainIndexer { return s.bloomIndexer }
|
||||
func (s *Ethereum) Merger() *consensus.Merger { return s.merger }
|
||||
|
||||
// Protocols returns all the currently configured
|
||||
// network protocols to start.
|
||||
|
|
|
|||
|
|
@ -267,12 +267,6 @@ func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payl
|
|||
finalized := api.remoteBlocks.get(update.FinalizedBlockHash)
|
||||
|
||||
// Header advertised via a past newPayload request. Start syncing to it.
|
||||
// Before we do however, make sure any legacy sync in switched off so we
|
||||
// don't accidentally have 2 cycles running.
|
||||
if merger := api.eth.Merger(); !merger.TDDReached() {
|
||||
merger.ReachTTD()
|
||||
api.eth.Downloader().Cancel()
|
||||
}
|
||||
context := []interface{}{"number", header.Number, "hash", header.Hash()}
|
||||
if update.FinalizedBlockHash != (common.Hash{}) {
|
||||
if finalized == nil {
|
||||
|
|
@ -334,9 +328,6 @@ func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payl
|
|||
// If the beacon client also advertised a finalized block, mark the local
|
||||
// chain final and completely in PoS mode.
|
||||
if update.FinalizedBlockHash != (common.Hash{}) {
|
||||
if merger := api.eth.Merger(); !merger.PoSFinalized() {
|
||||
merger.FinalizePoS()
|
||||
}
|
||||
// If the finalized block is not in our canonical tree, something is wrong
|
||||
finalBlock := api.eth.BlockChain().GetBlockByHash(update.FinalizedBlockHash)
|
||||
if finalBlock == nil {
|
||||
|
|
@ -620,13 +611,6 @@ func (api *ConsensusAPI) newPayload(params engine.ExecutableData, versionedHashe
|
|||
|
||||
return api.invalid(err, parent.Header()), nil
|
||||
}
|
||||
// We've accepted a valid payload from the beacon client. Mark the local
|
||||
// chain transitions to notify other subsystems (e.g. downloader) of the
|
||||
// behavioral change.
|
||||
if merger := api.eth.Merger(); !merger.TDDReached() {
|
||||
merger.ReachTTD()
|
||||
api.eth.Downloader().Cancel()
|
||||
}
|
||||
hash := block.Hash()
|
||||
return engine.PayloadStatusV1{Status: engine.VALID, LatestValidHash: &hash}, nil
|
||||
}
|
||||
|
|
@ -784,12 +768,10 @@ func (api *ConsensusAPI) heartbeat() {
|
|||
|
||||
// If there have been no updates for the past while, warn the user
|
||||
// that the beacon client is probably offline
|
||||
if api.eth.BlockChain().Config().TerminalTotalDifficultyPassed || api.eth.Merger().TDDReached() {
|
||||
if time.Since(lastForkchoiceUpdate) <= beaconUpdateConsensusTimeout || time.Since(lastNewPayloadUpdate) <= beaconUpdateConsensusTimeout {
|
||||
offlineLogged = time.Time{}
|
||||
continue
|
||||
}
|
||||
|
||||
if time.Since(offlineLogged) > beaconUpdateWarnFrequency {
|
||||
if lastForkchoiceUpdate.IsZero() && lastNewPayloadUpdate.IsZero() {
|
||||
if lastTransitionUpdate.IsZero() {
|
||||
|
|
@ -805,7 +787,6 @@ func (api *ConsensusAPI) heartbeat() {
|
|||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ExchangeCapabilities returns the current methods provided by this node.
|
||||
func (api *ConsensusAPI) ExchangeCapabilities([]string) []string {
|
||||
|
|
|
|||
|
|
@ -862,7 +862,6 @@ func TestTrickRemoteBlockCache(t *testing.T) {
|
|||
func TestInvalidBloom(t *testing.T) {
|
||||
genesis, preMergeBlocks := generateMergeChain(10, false)
|
||||
n, ethservice := startEthService(t, genesis, preMergeBlocks)
|
||||
ethservice.Merger().ReachTTD()
|
||||
defer n.Close()
|
||||
|
||||
commonAncestor := ethservice.BlockChain().CurrentBlock()
|
||||
|
|
@ -1044,7 +1043,6 @@ func TestWithdrawals(t *testing.T) {
|
|||
genesis.Config.ShanghaiTime = &time
|
||||
|
||||
n, ethservice := startEthService(t, genesis, blocks)
|
||||
ethservice.Merger().ReachTTD()
|
||||
defer n.Close()
|
||||
|
||||
api := NewConsensusAPI(ethservice)
|
||||
|
|
@ -1162,7 +1160,6 @@ func TestNilWithdrawals(t *testing.T) {
|
|||
genesis.Config.ShanghaiTime = &time
|
||||
|
||||
n, ethservice := startEthService(t, genesis, blocks)
|
||||
ethservice.Merger().ReachTTD()
|
||||
defer n.Close()
|
||||
|
||||
api := NewConsensusAPI(ethservice)
|
||||
|
|
@ -1589,7 +1586,6 @@ func TestParentBeaconBlockRoot(t *testing.T) {
|
|||
genesis.Config.CancunTime = &time
|
||||
|
||||
n, ethservice := startEthService(t, genesis, blocks)
|
||||
ethservice.Merger().ReachTTD()
|
||||
defer n.Close()
|
||||
|
||||
api := NewConsensusAPI(ethservice)
|
||||
|
|
|
|||
|
|
@ -165,15 +165,14 @@ type Config struct {
|
|||
// Clique is allowed for now to live standalone, but ethash is forbidden and can
|
||||
// only exist on already merged networks.
|
||||
func CreateConsensusEngine(config *params.ChainConfig, db ethdb.Database) (consensus.Engine, error) {
|
||||
// If proof-of-authority is requested, set it up
|
||||
// Geth v1.14.0 dropped support for non-merged networks in any consensus
|
||||
// mode. If such a network is requested, reject startup.
|
||||
if !config.TerminalTotalDifficultyPassed {
|
||||
return nil, errors.New("only PoS networks are supported, please transition old ones with Geth v1.13.x")
|
||||
}
|
||||
// Wrap previously supported consensus engines into their post-merge counterpart
|
||||
if config.Clique != nil {
|
||||
return beacon.New(clique.New(config.Clique, db)), nil
|
||||
}
|
||||
// If defaulting to proof-of-work, enforce an already merged network since
|
||||
// we cannot run PoW algorithms anymore, so we cannot even follow a chain
|
||||
// not coordinated by a beacon node.
|
||||
if !config.TerminalTotalDifficultyPassed {
|
||||
return nil, errors.New("ethash is only supported as a historical component of already merged networks")
|
||||
}
|
||||
return beacon.New(ethash.NewFaker()), nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,8 +25,6 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/consensus"
|
||||
"github.com/ethereum/go-ethereum/consensus/beacon"
|
||||
"github.com/ethereum/go-ethereum/core"
|
||||
"github.com/ethereum/go-ethereum/core/forkid"
|
||||
"github.com/ethereum/go-ethereum/core/rawdb"
|
||||
|
|
@ -91,7 +89,6 @@ type handlerConfig struct {
|
|||
Database ethdb.Database // Database for direct sync insertions
|
||||
Chain *core.BlockChain // Blockchain to serve data from
|
||||
TxPool txPool // Transaction pool to propagate from
|
||||
Merger *consensus.Merger // The manager for eth1/2 transition
|
||||
Network uint64 // Network identifier to advertise
|
||||
Sync downloader.SyncMode // Whether to snap or full sync
|
||||
BloomCache uint64 // Megabytes to alloc for snap sync bloom
|
||||
|
|
@ -115,12 +112,10 @@ type handler struct {
|
|||
downloader *downloader.Downloader
|
||||
txFetcher *fetcher.TxFetcher
|
||||
peers *peerSet
|
||||
merger *consensus.Merger
|
||||
|
||||
eventMux *event.TypeMux
|
||||
txsCh chan core.NewTxsEvent
|
||||
txsSub event.Subscription
|
||||
minedBlockSub *event.TypeMuxSubscription
|
||||
|
||||
requiredBlocks map[uint64]common.Hash
|
||||
|
||||
|
|
@ -148,7 +143,6 @@ func newHandler(config *handlerConfig) (*handler, error) {
|
|||
txpool: config.TxPool,
|
||||
chain: config.Chain,
|
||||
peers: newPeerSet(),
|
||||
merger: config.Merger,
|
||||
requiredBlocks: config.RequiredBlocks,
|
||||
quitSync: make(chan struct{}),
|
||||
handlerDoneCh: make(chan struct{}),
|
||||
|
|
@ -448,11 +442,6 @@ func (h *handler) Start(maxPeers int) {
|
|||
h.txsSub = h.txpool.SubscribeTransactions(h.txsCh, false)
|
||||
go h.txBroadcastLoop()
|
||||
|
||||
// broadcast mined blocks
|
||||
h.wg.Add(1)
|
||||
h.minedBlockSub = h.eventMux.Subscribe(core.NewMinedBlockEvent{})
|
||||
go h.minedBroadcastLoop()
|
||||
|
||||
// start peer handler tracker
|
||||
h.wg.Add(1)
|
||||
go h.protoTracker()
|
||||
|
|
@ -460,7 +449,6 @@ func (h *handler) Start(maxPeers int) {
|
|||
|
||||
func (h *handler) Stop() {
|
||||
h.txsSub.Unsubscribe() // quits txBroadcastLoop
|
||||
h.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
|
||||
|
||||
// Quit chainSync and txsync64.
|
||||
// After this is done, no new peers will be accepted.
|
||||
|
|
@ -476,50 +464,6 @@ func (h *handler) Stop() {
|
|||
log.Info("Ethereum protocol stopped")
|
||||
}
|
||||
|
||||
// BroadcastBlock will either propagate a block to a subset of its peers, or
|
||||
// will only announce its availability (depending what's requested).
|
||||
func (h *handler) BroadcastBlock(block *types.Block, propagate bool) {
|
||||
// Disable the block propagation if the chain has already entered the PoS
|
||||
// stage. The block propagation is delegated to the consensus layer.
|
||||
if h.merger.PoSFinalized() {
|
||||
return
|
||||
}
|
||||
// Disable the block propagation if it's the post-merge block.
|
||||
if beacon, ok := h.chain.Engine().(*beacon.Beacon); ok {
|
||||
if beacon.IsPoSHeader(block.Header()) {
|
||||
return
|
||||
}
|
||||
}
|
||||
hash := block.Hash()
|
||||
peers := h.peers.peersWithoutBlock(hash)
|
||||
|
||||
// If propagation is requested, send to a subset of the peer
|
||||
if propagate {
|
||||
// Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
|
||||
var td *big.Int
|
||||
if parent := h.chain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
|
||||
td = new(big.Int).Add(block.Difficulty(), h.chain.GetTd(block.ParentHash(), block.NumberU64()-1))
|
||||
} else {
|
||||
log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
|
||||
return
|
||||
}
|
||||
// Send the block to a subset of our peers
|
||||
transfer := peers[:int(math.Sqrt(float64(len(peers))))]
|
||||
for _, peer := range transfer {
|
||||
peer.AsyncSendNewBlock(block, td)
|
||||
}
|
||||
log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
|
||||
return
|
||||
}
|
||||
// Otherwise if the block is indeed in out own chain, announce it
|
||||
if h.chain.HasBlock(hash, block.NumberU64()) {
|
||||
for _, peer := range peers {
|
||||
peer.AsyncSendNewBlockHash(block)
|
||||
}
|
||||
log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
|
||||
}
|
||||
}
|
||||
|
||||
// BroadcastTransactions will propagate a batch of transactions
|
||||
// - To a square root of all peers for non-blob transactions
|
||||
// - And, separately, as announcements to all peers which are not known to
|
||||
|
|
@ -602,18 +546,6 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
|
|||
"bcastpeers", directPeers, "bcastcount", directCount, "annpeers", annPeers, "anncount", annCount)
|
||||
}
|
||||
|
||||
// minedBroadcastLoop sends mined blocks to connected peers.
|
||||
func (h *handler) minedBroadcastLoop() {
|
||||
defer h.wg.Done()
|
||||
|
||||
for obj := range h.minedBlockSub.Chan() {
|
||||
if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
|
||||
h.BroadcastBlock(ev.Block, true) // First propagate block to peers
|
||||
h.BroadcastBlock(ev.Block, false) // Only then announce to the rest
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// txBroadcastLoop announces new transactions to connected peers.
|
||||
func (h *handler) txBroadcastLoop() {
|
||||
defer h.wg.Done()
|
||||
|
|
|
|||
|
|
@ -57,12 +57,6 @@ func (h *ethHandler) AcceptTxs() bool {
|
|||
func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
|
||||
// Consume any broadcasts and announces, forwarding the rest to the downloader
|
||||
switch packet := packet.(type) {
|
||||
case *eth.NewBlockHashesPacket:
|
||||
return errors.New("block announcements disallowed")
|
||||
|
||||
case *eth.NewBlockPacket:
|
||||
return errors.New("block broadcasts disallowed")
|
||||
|
||||
case *eth.NewPooledTransactionHashesPacket:
|
||||
return h.txFetcher.Notify(peer.ID(), packet.Types, packet.Sizes, packet.Hashes)
|
||||
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/consensus"
|
||||
"github.com/ethereum/go-ethereum/consensus/ethash"
|
||||
"github.com/ethereum/go-ethereum/core"
|
||||
"github.com/ethereum/go-ethereum/core/forkid"
|
||||
|
|
@ -109,7 +108,6 @@ func testForkIDSplit(t *testing.T, protocol uint) {
|
|||
Database: dbNoFork,
|
||||
Chain: chainNoFork,
|
||||
TxPool: newTestTxPool(),
|
||||
Merger: consensus.NewMerger(rawdb.NewMemoryDatabase()),
|
||||
Network: 1,
|
||||
Sync: downloader.FullSync,
|
||||
BloomCache: 1,
|
||||
|
|
@ -118,7 +116,6 @@ func testForkIDSplit(t *testing.T, protocol uint) {
|
|||
Database: dbProFork,
|
||||
Chain: chainProFork,
|
||||
TxPool: newTestTxPool(),
|
||||
Merger: consensus.NewMerger(rawdb.NewMemoryDatabase()),
|
||||
Network: 1,
|
||||
Sync: downloader.FullSync,
|
||||
BloomCache: 1,
|
||||
|
|
@ -441,159 +438,3 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tests that blocks are broadcast to a sqrt number of peers only.
|
||||
func TestBroadcastBlock1Peer(t *testing.T) { testBroadcastBlock(t, 1, 1) }
|
||||
func TestBroadcastBlock2Peers(t *testing.T) { testBroadcastBlock(t, 2, 1) }
|
||||
func TestBroadcastBlock3Peers(t *testing.T) { testBroadcastBlock(t, 3, 1) }
|
||||
func TestBroadcastBlock4Peers(t *testing.T) { testBroadcastBlock(t, 4, 2) }
|
||||
func TestBroadcastBlock5Peers(t *testing.T) { testBroadcastBlock(t, 5, 2) }
|
||||
func TestBroadcastBlock8Peers(t *testing.T) { testBroadcastBlock(t, 9, 3) }
|
||||
func TestBroadcastBlock12Peers(t *testing.T) { testBroadcastBlock(t, 12, 3) }
|
||||
func TestBroadcastBlock16Peers(t *testing.T) { testBroadcastBlock(t, 16, 4) }
|
||||
func TestBroadcastBloc26Peers(t *testing.T) { testBroadcastBlock(t, 26, 5) }
|
||||
func TestBroadcastBlock100Peers(t *testing.T) { testBroadcastBlock(t, 100, 10) }
|
||||
|
||||
func testBroadcastBlock(t *testing.T, peers, bcasts int) {
|
||||
t.Parallel()
|
||||
|
||||
// Create a source handler to broadcast blocks from and a number of sinks
|
||||
// to receive them.
|
||||
source := newTestHandlerWithBlocks(1)
|
||||
defer source.close()
|
||||
|
||||
sinks := make([]*testEthHandler, peers)
|
||||
for i := 0; i < len(sinks); i++ {
|
||||
sinks[i] = new(testEthHandler)
|
||||
}
|
||||
// Interconnect all the sink handlers with the source handler
|
||||
var (
|
||||
genesis = source.chain.Genesis()
|
||||
td = source.chain.GetTd(genesis.Hash(), genesis.NumberU64())
|
||||
)
|
||||
for i, sink := range sinks {
|
||||
sink := sink // Closure for gorotuine below
|
||||
|
||||
sourcePipe, sinkPipe := p2p.MsgPipe()
|
||||
defer sourcePipe.Close()
|
||||
defer sinkPipe.Close()
|
||||
|
||||
sourcePeer := eth.NewPeer(eth.ETH68, p2p.NewPeerPipe(enode.ID{byte(i)}, "", nil, sourcePipe), sourcePipe, nil)
|
||||
sinkPeer := eth.NewPeer(eth.ETH68, p2p.NewPeerPipe(enode.ID{0}, "", nil, sinkPipe), sinkPipe, nil)
|
||||
defer sourcePeer.Close()
|
||||
defer sinkPeer.Close()
|
||||
|
||||
go source.handler.runEthPeer(sourcePeer, func(peer *eth.Peer) error {
|
||||
return eth.Handle((*ethHandler)(source.handler), peer)
|
||||
})
|
||||
if err := sinkPeer.Handshake(1, td, genesis.Hash(), genesis.Hash(), forkid.NewIDWithChain(source.chain), forkid.NewFilter(source.chain)); err != nil {
|
||||
t.Fatalf("failed to run protocol handshake")
|
||||
}
|
||||
go eth.Handle(sink, sinkPeer)
|
||||
}
|
||||
// Subscribe to all the transaction pools
|
||||
blockChs := make([]chan *types.Block, len(sinks))
|
||||
for i := 0; i < len(sinks); i++ {
|
||||
blockChs[i] = make(chan *types.Block, 1)
|
||||
defer close(blockChs[i])
|
||||
|
||||
sub := sinks[i].blockBroadcasts.Subscribe(blockChs[i])
|
||||
defer sub.Unsubscribe()
|
||||
}
|
||||
// Initiate a block propagation across the peers
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
header := source.chain.CurrentBlock()
|
||||
source.handler.BroadcastBlock(source.chain.GetBlock(header.Hash(), header.Number.Uint64()), true)
|
||||
|
||||
// Iterate through all the sinks and ensure the correct number got the block
|
||||
done := make(chan struct{}, peers)
|
||||
for _, ch := range blockChs {
|
||||
ch := ch
|
||||
go func() {
|
||||
<-ch
|
||||
done <- struct{}{}
|
||||
}()
|
||||
}
|
||||
var received int
|
||||
for {
|
||||
select {
|
||||
case <-done:
|
||||
received++
|
||||
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
if received != bcasts {
|
||||
t.Errorf("broadcast count mismatch: have %d, want %d", received, bcasts)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tests that a propagated malformed block (uncles or transactions don't match
|
||||
// with the hashes in the header) gets discarded and not broadcast forward.
|
||||
func TestBroadcastMalformedBlock68(t *testing.T) { testBroadcastMalformedBlock(t, eth.ETH68) }
|
||||
|
||||
func testBroadcastMalformedBlock(t *testing.T, protocol uint) {
|
||||
t.Parallel()
|
||||
|
||||
// Create a source handler to broadcast blocks from and a number of sinks
|
||||
// to receive them.
|
||||
source := newTestHandlerWithBlocks(1)
|
||||
defer source.close()
|
||||
|
||||
// Create a source handler to send messages through and a sink peer to receive them
|
||||
p2pSrc, p2pSink := p2p.MsgPipe()
|
||||
defer p2pSrc.Close()
|
||||
defer p2pSink.Close()
|
||||
|
||||
src := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{1}, "", nil, p2pSrc), p2pSrc, source.txpool)
|
||||
sink := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{2}, "", nil, p2pSink), p2pSink, source.txpool)
|
||||
defer src.Close()
|
||||
defer sink.Close()
|
||||
|
||||
go source.handler.runEthPeer(src, func(peer *eth.Peer) error {
|
||||
return eth.Handle((*ethHandler)(source.handler), peer)
|
||||
})
|
||||
// Run the handshake locally to avoid spinning up a sink handler
|
||||
var (
|
||||
genesis = source.chain.Genesis()
|
||||
td = source.chain.GetTd(genesis.Hash(), genesis.NumberU64())
|
||||
)
|
||||
if err := sink.Handshake(1, td, genesis.Hash(), genesis.Hash(), forkid.NewIDWithChain(source.chain), forkid.NewFilter(source.chain)); err != nil {
|
||||
t.Fatalf("failed to run protocol handshake")
|
||||
}
|
||||
// After the handshake completes, the source handler should stream the sink
|
||||
// the blocks, subscribe to inbound network events
|
||||
backend := new(testEthHandler)
|
||||
|
||||
blocks := make(chan *types.Block, 1)
|
||||
sub := backend.blockBroadcasts.Subscribe(blocks)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
go eth.Handle(backend, sink)
|
||||
|
||||
// Create various combinations of malformed blocks
|
||||
head := source.chain.CurrentBlock()
|
||||
block := source.chain.GetBlock(head.Hash(), head.Number.Uint64())
|
||||
|
||||
malformedUncles := head
|
||||
malformedUncles.UncleHash[0]++
|
||||
malformedTransactions := head
|
||||
malformedTransactions.TxHash[0]++
|
||||
malformedEverything := head
|
||||
malformedEverything.UncleHash[0]++
|
||||
malformedEverything.TxHash[0]++
|
||||
|
||||
// Try to broadcast all malformations and ensure they all get discarded
|
||||
for _, header := range []*types.Header{malformedUncles, malformedTransactions, malformedEverything} {
|
||||
block := types.NewBlockWithHeader(header).WithBody(block.Transactions(), block.Uncles())
|
||||
if err := src.SendNewBlock(block, big.NewInt(131136)); err != nil {
|
||||
t.Fatalf("failed to broadcast block: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-blocks:
|
||||
t.Fatalf("malformed block forwarded")
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,7 +22,6 @@ import (
|
|||
"sync"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/consensus"
|
||||
"github.com/ethereum/go-ethereum/consensus/ethash"
|
||||
"github.com/ethereum/go-ethereum/core"
|
||||
"github.com/ethereum/go-ethereum/core/rawdb"
|
||||
|
|
@ -164,7 +163,6 @@ func newTestHandlerWithBlocks(blocks int) *testHandler {
|
|||
Database: db,
|
||||
Chain: chain,
|
||||
TxPool: txpool,
|
||||
Merger: consensus.NewMerger(rawdb.NewMemoryDatabase()),
|
||||
Network: 1,
|
||||
Sync: downloader.SnapSync,
|
||||
BloomCache: 1,
|
||||
|
|
|
|||
|
|
@ -192,21 +192,6 @@ func (ps *peerSet) peer(id string) *ethPeer {
|
|||
return ps.peers[id]
|
||||
}
|
||||
|
||||
// peersWithoutBlock retrieves a list of peers that do not have a given block in
|
||||
// their set of known hashes so it might be propagated to them.
|
||||
func (ps *peerSet) peersWithoutBlock(hash common.Hash) []*ethPeer {
|
||||
ps.lock.RLock()
|
||||
defer ps.lock.RUnlock()
|
||||
|
||||
list := make([]*ethPeer, 0, len(ps.peers))
|
||||
for _, p := range ps.peers {
|
||||
if !p.KnownBlock(hash) {
|
||||
list = append(list, p)
|
||||
}
|
||||
}
|
||||
return list
|
||||
}
|
||||
|
||||
// peersWithoutTransaction retrieves a list of peers that do not have a given
|
||||
// transaction in their set of known hashes.
|
||||
func (ps *peerSet) peersWithoutTransaction(hash common.Hash) []*ethPeer {
|
||||
|
|
|
|||
|
|
@ -36,30 +36,6 @@ type blockPropagation struct {
|
|||
td *big.Int
|
||||
}
|
||||
|
||||
// broadcastBlocks is a write loop that multiplexes blocks and block announcements
|
||||
// to the remote peer. The goal is to have an async writer that does not lock up
|
||||
// node internals and at the same time rate limits queued data.
|
||||
func (p *Peer) broadcastBlocks() {
|
||||
for {
|
||||
select {
|
||||
case prop := <-p.queuedBlocks:
|
||||
if err := p.SendNewBlock(prop.block, prop.td); err != nil {
|
||||
return
|
||||
}
|
||||
p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)
|
||||
|
||||
case block := <-p.queuedBlockAnns:
|
||||
if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil {
|
||||
return
|
||||
}
|
||||
p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())
|
||||
|
||||
case <-p.term:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// broadcastTransactions is a write loop that schedules transaction broadcasts
|
||||
// to the remote peer. The goal is to have an async writer that does not lock up
|
||||
// node internals and at the same time rate limits queued data.
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ package eth
|
|||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
|
|
@ -274,43 +275,11 @@ func ServiceGetReceiptsQuery(chain *core.BlockChain, query GetReceiptsRequest) [
|
|||
}
|
||||
|
||||
func handleNewBlockhashes(backend Backend, msg Decoder, peer *Peer) error {
|
||||
// A batch of new block announcements just arrived
|
||||
ann := new(NewBlockHashesPacket)
|
||||
if err := msg.Decode(ann); err != nil {
|
||||
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
|
||||
}
|
||||
// Mark the hashes as present at the remote node
|
||||
for _, block := range *ann {
|
||||
peer.markBlock(block.Hash)
|
||||
}
|
||||
// Deliver them all to the backend for queuing
|
||||
return backend.Handle(peer, ann)
|
||||
return errors.New("block announcements disallowed") // We dropped support for non-merge networks
|
||||
}
|
||||
|
||||
func handleNewBlock(backend Backend, msg Decoder, peer *Peer) error {
|
||||
// Retrieve and decode the propagated block
|
||||
ann := new(NewBlockPacket)
|
||||
if err := msg.Decode(ann); err != nil {
|
||||
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
|
||||
}
|
||||
if err := ann.sanityCheck(); err != nil {
|
||||
return err
|
||||
}
|
||||
if hash := types.CalcUncleHash(ann.Block.Uncles()); hash != ann.Block.UncleHash() {
|
||||
log.Warn("Propagated block has invalid uncles", "have", hash, "exp", ann.Block.UncleHash())
|
||||
return nil // TODO(karalabe): return error eventually, but wait a few releases
|
||||
}
|
||||
if hash := types.DeriveSha(ann.Block.Transactions(), trie.NewStackTrie(nil)); hash != ann.Block.TxHash() {
|
||||
log.Warn("Propagated block has invalid body", "have", hash, "exp", ann.Block.TxHash())
|
||||
return nil // TODO(karalabe): return error eventually, but wait a few releases
|
||||
}
|
||||
ann.Block.ReceivedAt = msg.Time()
|
||||
ann.Block.ReceivedFrom = peer
|
||||
|
||||
// Mark the peer as owning the block
|
||||
peer.markBlock(ann.Block.Hash())
|
||||
|
||||
return backend.Handle(peer, ann)
|
||||
return errors.New("block broadcasts disallowed") // We dropped support for non-merge networks
|
||||
}
|
||||
|
||||
func handleBlockHeaders(backend Backend, msg Decoder, peer *Peer) error {
|
||||
|
|
|
|||
|
|
@ -33,10 +33,6 @@ const (
|
|||
// before starting to randomly evict them.
|
||||
maxKnownTxs = 32768
|
||||
|
||||
// maxKnownBlocks is the maximum block hashes to keep in the known list
|
||||
// before starting to randomly evict them.
|
||||
maxKnownBlocks = 1024
|
||||
|
||||
// maxQueuedTxs is the maximum number of transactions to queue up before dropping
|
||||
// older broadcasts.
|
||||
maxQueuedTxs = 4096
|
||||
|
|
@ -44,16 +40,6 @@ const (
|
|||
// maxQueuedTxAnns is the maximum number of transaction announcements to queue up
|
||||
// before dropping older announcements.
|
||||
maxQueuedTxAnns = 4096
|
||||
|
||||
// maxQueuedBlocks is the maximum number of block propagations to queue up before
|
||||
// dropping broadcasts. There's not much point in queueing stale blocks, so a few
|
||||
// that might cover uncles should be enough.
|
||||
maxQueuedBlocks = 4
|
||||
|
||||
// maxQueuedBlockAnns is the maximum number of block announcements to queue up before
|
||||
// dropping broadcasts. Similarly to block propagations, there's no point to queue
|
||||
// above some healthy uncle limit, so use that.
|
||||
maxQueuedBlockAnns = 4
|
||||
)
|
||||
|
||||
// max is a helper function which returns the larger of the two given integers.
|
||||
|
|
@ -75,10 +61,6 @@ type Peer struct {
|
|||
head common.Hash // Latest advertised head block hash
|
||||
td *big.Int // Latest advertised head block total difficulty
|
||||
|
||||
knownBlocks *knownCache // Set of block hashes known to be known by this peer
|
||||
queuedBlocks chan *blockPropagation // Queue of blocks to broadcast to the peer
|
||||
queuedBlockAnns chan *types.Block // Queue of blocks to announce to the peer
|
||||
|
||||
txpool TxPool // Transaction pool used by the broadcasters for liveness checks
|
||||
knownTxs *knownCache // Set of transaction hashes known to be known by this peer
|
||||
txBroadcast chan []common.Hash // Channel used to queue transaction propagation requests
|
||||
|
|
@ -101,9 +83,6 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Pe
|
|||
rw: rw,
|
||||
version: version,
|
||||
knownTxs: newKnownCache(maxKnownTxs),
|
||||
knownBlocks: newKnownCache(maxKnownBlocks),
|
||||
queuedBlocks: make(chan *blockPropagation, maxQueuedBlocks),
|
||||
queuedBlockAnns: make(chan *types.Block, maxQueuedBlockAnns),
|
||||
txBroadcast: make(chan []common.Hash),
|
||||
txAnnounce: make(chan []common.Hash),
|
||||
reqDispatch: make(chan *request),
|
||||
|
|
@ -113,7 +92,6 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Pe
|
|||
term: make(chan struct{}),
|
||||
}
|
||||
// Start up all the broadcasters
|
||||
go peer.broadcastBlocks()
|
||||
go peer.broadcastTransactions()
|
||||
go peer.announceTransactions()
|
||||
go peer.dispatcher()
|
||||
|
|
@ -156,23 +134,11 @@ func (p *Peer) SetHead(hash common.Hash, td *big.Int) {
|
|||
p.td.Set(td)
|
||||
}
|
||||
|
||||
// KnownBlock returns whether peer is known to already have a block.
|
||||
func (p *Peer) KnownBlock(hash common.Hash) bool {
|
||||
return p.knownBlocks.Contains(hash)
|
||||
}
|
||||
|
||||
// KnownTransaction returns whether peer is known to already have a transaction.
|
||||
func (p *Peer) KnownTransaction(hash common.Hash) bool {
|
||||
return p.knownTxs.Contains(hash)
|
||||
}
|
||||
|
||||
// markBlock marks a block as known for the peer, ensuring that the block will
|
||||
// never be propagated to this particular peer.
|
||||
func (p *Peer) markBlock(hash common.Hash) {
|
||||
// If we reached the memory allowance, drop a previously known block hash
|
||||
p.knownBlocks.Add(hash)
|
||||
}
|
||||
|
||||
// markTransaction marks a transaction as known for the peer, ensuring that it
|
||||
// will never be propagated to this particular peer.
|
||||
func (p *Peer) markTransaction(hash common.Hash) {
|
||||
|
|
@ -248,55 +214,6 @@ func (p *Peer) ReplyPooledTransactionsRLP(id uint64, hashes []common.Hash, txs [
|
|||
})
|
||||
}
|
||||
|
||||
// SendNewBlockHashes announces the availability of a number of blocks through
|
||||
// a hash notification.
|
||||
func (p *Peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error {
|
||||
// Mark all the block hashes as known, but ensure we don't overflow our limits
|
||||
p.knownBlocks.Add(hashes...)
|
||||
|
||||
request := make(NewBlockHashesPacket, len(hashes))
|
||||
for i := 0; i < len(hashes); i++ {
|
||||
request[i].Hash = hashes[i]
|
||||
request[i].Number = numbers[i]
|
||||
}
|
||||
return p2p.Send(p.rw, NewBlockHashesMsg, request)
|
||||
}
|
||||
|
||||
// AsyncSendNewBlockHash queues the availability of a block for propagation to a
|
||||
// remote peer. If the peer's broadcast queue is full, the event is silently
|
||||
// dropped.
|
||||
func (p *Peer) AsyncSendNewBlockHash(block *types.Block) {
|
||||
select {
|
||||
case p.queuedBlockAnns <- block:
|
||||
// Mark all the block hash as known, but ensure we don't overflow our limits
|
||||
p.knownBlocks.Add(block.Hash())
|
||||
default:
|
||||
p.Log().Debug("Dropping block announcement", "number", block.NumberU64(), "hash", block.Hash())
|
||||
}
|
||||
}
|
||||
|
||||
// SendNewBlock propagates an entire block to a remote peer.
|
||||
func (p *Peer) SendNewBlock(block *types.Block, td *big.Int) error {
|
||||
// Mark all the block hash as known, but ensure we don't overflow our limits
|
||||
p.knownBlocks.Add(block.Hash())
|
||||
return p2p.Send(p.rw, NewBlockMsg, &NewBlockPacket{
|
||||
Block: block,
|
||||
TD: td,
|
||||
})
|
||||
}
|
||||
|
||||
// AsyncSendNewBlock queues an entire block for propagation to a remote peer. If
|
||||
// the peer's broadcast queue is full, the event is silently dropped.
|
||||
func (p *Peer) AsyncSendNewBlock(block *types.Block, td *big.Int) {
|
||||
select {
|
||||
case p.queuedBlocks <- &blockPropagation{block: block, td: td}:
|
||||
// Mark all the block hash as known, but ensure we don't overflow our limits
|
||||
p.knownBlocks.Add(block.Hash())
|
||||
default:
|
||||
p.Log().Debug("Dropping block propagation", "number", block.NumberU64(), "hash", block.Hash())
|
||||
}
|
||||
}
|
||||
|
||||
// ReplyBlockHeadersRLP is the response to GetBlockHeaders.
|
||||
func (p *Peer) ReplyBlockHeadersRLP(id uint64, headers []rlp.RawValue) error {
|
||||
return p2p.Send(p.rw, BlockHeadersMsg, &BlockHeadersRLPPacket{
|
||||
|
|
|
|||
|
|
@ -85,10 +85,11 @@ func testSnapSyncDisabling(t *testing.T, ethVer uint, snapVer uint) {
|
|||
time.Sleep(250 * time.Millisecond)
|
||||
|
||||
// Check that snap sync was disabled
|
||||
op := peerToSyncOp(downloader.SnapSync, empty.handler.peers.peerWithHighestTD())
|
||||
if err := empty.handler.doSync(op); err != nil {
|
||||
if err := empty.handler.downloader.BeaconSync(downloader.SnapSync, full.chain.CurrentBlock(), nil); err != nil {
|
||||
t.Fatal("sync failed:", err)
|
||||
}
|
||||
empty.handler.enableSyncedFeatures()
|
||||
|
||||
if empty.handler.snapSync.Load() {
|
||||
t.Fatalf("snap sync not disabled after successful synchronisation")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -361,6 +361,8 @@ type ChainConfig struct {
|
|||
// TerminalTotalDifficultyPassed is a flag specifying that the network already
|
||||
// passed the terminal total difficulty. Its purpose is to disable legacy sync
|
||||
// even without having seen the TTD locally (safer long term).
|
||||
//
|
||||
// TODO(karalabe): Drop this field eventually (always assuming PoS mode)
|
||||
TerminalTotalDifficultyPassed bool `json:"terminalTotalDifficultyPassed,omitempty"`
|
||||
|
||||
// Various consensus engines
|
||||
|
|
|
|||
Loading…
Reference in a new issue