Revert "consensus, cmd, core, eth: remove support for non-merge mode of operation (#29169)"

This reverts commit a6eb14b.

Author: buddh0

cmd/geth/testdata/clique.json

@@ -8,7 +8,6 @@
     "byzantiumBlock": 0,
     "constantinopleBlock": 0,
     "petersburgBlock": 0,
-    "terminalTotalDifficultyPassed": true,
     "clique": {
       "period": 5,
       "epoch": 30000

consensus/merger.go

@@ -0,0 +1,110 @@
+// 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
+}

core/block_validator_test.go

@@ -94,6 +94,7 @@ func testHeaderVerificationForMerging(t *testing.T, isClique bool) {
 		preBlocks  []*types.Block
 		postBlocks []*types.Block
 		engine     consensus.Engine
+		merger     = consensus.NewMerger(rawdb.NewMemoryDatabase())
 	)
 	if isClique {
 		var (
@@ -185,6 +186,11 @@ func testHeaderVerificationForMerging(t *testing.T, isClique bool) {
 		}
 		chain.InsertChain(preBlocks[i : i+1])
 	}
+
+	// Make the transition
+	merger.ReachTTD()
+	merger.FinalizePoS()
+
 	// Verify the blocks after the merging
 	for i := 0; i < len(postBlocks); i++ {
 		_, results := engine.VerifyHeaders(chain, []*types.Header{postHeaders[i]})

core/blockchain.go

@@ -56,6 +56,7 @@ import (
 	"github.com/ethereum/go-ethereum/triedb"
 	"github.com/ethereum/go-ethereum/triedb/hashdb"
 	"github.com/ethereum/go-ethereum/triedb/pathdb"
+	"golang.org/x/exp/slices"
 )
 
 var (
@@ -109,7 +110,6 @@ var (
 )
 
 const (
-<<<<<<< HEAD
 	bodyCacheLimit      = 256
 	blockCacheLimit     = 256
 	diffLayerCacheLimit = 1024
@@ -127,13 +127,6 @@ const (
 	maxDiffForkDist                 = 11 // Maximum allowed backward distance from the chain head
 
 	rewindBadBlockInterval = 1 * time.Second
-=======
-	bodyCacheLimit     = 256
-	blockCacheLimit    = 256
-	receiptsCacheLimit = 32
-	txLookupCacheLimit = 1024
-	TriesInMemory      = 128
->>>>>>> f4d53133f (consensus, cmd, core, eth: remove support for non-merge mode of operation (#29169))
 
 	// BlockChainVersion ensures that an incompatible database forces a resync from scratch.
 	//
@@ -299,7 +292,6 @@ type BlockChain struct {
 	txLookupCache *lru.Cache[common.Hash, txLookup]
 	sidecarsCache *lru.Cache[common.Hash, types.BlobSidecars]
 
-<<<<<<< HEAD
 	// future blocks are blocks added for later processing
 	futureBlocks *lru.Cache[common.Hash, *types.Block]
 	// Cache for the blocks that failed to pass MPT root verification
@@ -312,8 +304,6 @@ type BlockChain struct {
 	diffQueueBuffer            chan *types.DiffLayer
 	diffLayerFreezerBlockLimit uint64
 
-=======
->>>>>>> f4d53133f (consensus, cmd, core, eth: remove support for non-merge mode of operation (#29169))
 	wg            sync.WaitGroup
 	dbWg          sync.WaitGroup
 	quit          chan struct{} // shutdown signal, closed in Stop.
@@ -373,7 +363,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 	*/
 
 	bc := &BlockChain{
-<<<<<<< HEAD
 		chainConfig:        chainConfig,
 		cacheConfig:        cacheConfig,
 		db:                 db,
@@ -396,22 +385,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 		vmConfig:           vmConfig,
 		diffQueue:          prque.New[int64, *types.DiffLayer](nil),
 		diffQueueBuffer:    make(chan *types.DiffLayer),
-=======
-		chainConfig:   chainConfig,
-		cacheConfig:   cacheConfig,
-		db:            db,
-		triedb:        triedb,
-		triegc:        prque.New[int64, common.Hash](nil),
-		quit:          make(chan struct{}),
-		chainmu:       syncx.NewClosableMutex(),
-		bodyCache:     lru.NewCache[common.Hash, *types.Body](bodyCacheLimit),
-		bodyRLPCache:  lru.NewCache[common.Hash, rlp.RawValue](bodyCacheLimit),
-		receiptsCache: lru.NewCache[common.Hash, []*types.Receipt](receiptsCacheLimit),
-		blockCache:    lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
-		txLookupCache: lru.NewCache[common.Hash, txLookup](txLookupCacheLimit),
-		engine:        engine,
-		vmConfig:      vmConfig,
->>>>>>> f4d53133f (consensus, cmd, core, eth: remove support for non-merge mode of operation (#29169))
 	}
 	bc.flushInterval.Store(int64(cacheConfig.TrieTimeLimit))
 	bc.forker = NewForkChoice(bc, shouldPreserve)
@@ -577,7 +550,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 			return nil, err
 		}
 	}
-<<<<<<< HEAD
 	// Start future block processor.
 	bc.wg.Add(1)
 	go bc.updateFutureBlocks()
@@ -598,8 +570,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 		go bc.startDoubleSignMonitor()
 	}
 
-=======
->>>>>>> f4d53133f (consensus, cmd, core, eth: remove support for non-merge mode of operation (#29169))
 	// Rewind the chain in case of an incompatible config upgrade.
 	if compat, ok := genesisErr.(*params.ConfigCompatError); ok {
 		log.Warn("Rewinding chain to upgrade configuration", "err", compat)
@@ -1187,6 +1157,7 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
 	bc.sidecarsCache.Purge()
 	bc.blockCache.Purge()
 	bc.txLookupCache.Purge()
+	bc.futureBlocks.Purge()
 
 	if finalized := bc.CurrentFinalBlock(); finalized != nil && head < finalized.Number.Uint64() {
 		log.Error("SetHead invalidated finalized block")
@@ -1474,6 +1445,24 @@ func (bc *BlockChain) insertStopped() bool {
 	return bc.procInterrupt.Load()
 }
 
+func (bc *BlockChain) procFutureBlocks() {
+	blocks := make([]*types.Block, 0, bc.futureBlocks.Len())
+	for _, hash := range bc.futureBlocks.Keys() {
+		if block, exist := bc.futureBlocks.Peek(hash); exist {
+			blocks = append(blocks, block)
+		}
+	}
+	if len(blocks) > 0 {
+		slices.SortFunc(blocks, func(a, b *types.Block) int {
+			return a.Number().Cmp(b.Number())
+		})
+		// Insert one by one as chain insertion needs contiguous ancestry between blocks
+		for i := range blocks {
+			bc.InsertChain(blocks[i : i+1])
+		}
+	}
+}
+
 // WriteStatus status of write
 type WriteStatus byte
 
@@ -1970,6 +1959,8 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 	if status == CanonStatTy {
 		bc.writeHeadBlock(block)
 	}
+	bc.futureBlocks.Remove(block.Hash())
+
 	if status == CanonStatTy {
 		bc.chainFeed.Send(ChainEvent{Block: block, Hash: block.Hash(), Logs: logs})
 		if len(logs) > 0 {
@@ -1998,6 +1989,25 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 	return status, nil
 }
 
+// addFutureBlock checks if the block is within the max allowed window to get
+// accepted for future processing, and returns an error if the block is too far
+// ahead and was not added.
+//
+// TODO after the transition, the future block shouldn't be kept. Because
+// it's not checked in the Geth side anymore.
+func (bc *BlockChain) addFutureBlock(block *types.Block) error {
+	max := uint64(time.Now().Unix() + maxTimeFutureBlocks)
+	if block.Time() > max {
+		return fmt.Errorf("future block timestamp %v > allowed %v", block.Time(), max)
+	}
+	if block.Difficulty().Cmp(common.Big0) == 0 {
+		// Never add PoS blocks into the future queue
+		return nil
+	}
+	bc.futureBlocks.Add(block.Hash(), block)
+	return nil
+}
+
 // InsertChain attempts to insert the given batch of blocks in to the canonical
 // chain or, otherwise, create a fork. If an error is returned it will return
 // the index number of the failing block as well an error describing what went
@@ -2152,10 +2162,26 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 			_, err := bc.recoverAncestors(block)
 			return it.index, err
 		}
+	// First block is future, shove it (and all children) to the future queue (unknown ancestor)
+	case errors.Is(err, consensus.ErrFutureBlock) || (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(it.first().ParentHash())):
+		for block != nil && (it.index == 0 || errors.Is(err, consensus.ErrUnknownAncestor)) {
+			log.Debug("Future block, postponing import", "number", block.Number(), "hash", block.Hash())
+			if err := bc.addFutureBlock(block); err != nil {
+				return it.index, err
+			}
+			block, err = it.next()
+		}
+		stats.queued += it.processed()
+		stats.ignored += it.remaining()
+
+		// If there are any still remaining, mark as ignored
+		return it.index, err
+
 	// Some other error(except ErrKnownBlock) occurred, abort.
 	// ErrKnownBlock is allowed here since some known blocks
 	// still need re-execution to generate snapshots that are missing
 	case err != nil && !errors.Is(err, ErrKnownBlock):
+		bc.futureBlocks.Remove(block.Hash())
 		stats.ignored += len(it.chain)
 		bc.reportBlock(block, nil, err)
 		return it.index, err
@@ -2354,7 +2380,23 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 		}
 		bc.chainBlockFeed.Send(ChainHeadEvent{block})
 	}
+
+	// Any blocks remaining here? The only ones we care about are the future ones
+	if block != nil && errors.Is(err, consensus.ErrFutureBlock) {
+		if err := bc.addFutureBlock(block); err != nil {
+			return it.index, err
+		}
+		block, err = it.next()
+
+		for ; block != nil && errors.Is(err, consensus.ErrUnknownAncestor); block, err = it.next() {
+			if err := bc.addFutureBlock(block); err != nil {
+				return it.index, err
+			}
+			stats.queued++
+		}
+	}
 	stats.ignored += it.remaining()
+
 	return it.index, err
 }
 
@@ -2833,7 +2875,6 @@ func (bc *BlockChain) SetCanonical(head *types.Block) (common.Hash, error) {
 	return head.Hash(), nil
 }
 
-<<<<<<< HEAD
 func (bc *BlockChain) updateFutureBlocks() {
 	futureTimer := time.NewTicker(5 * time.Second)
 	defer futureTimer.Stop()
@@ -2960,8 +3001,6 @@ func (bc *BlockChain) startDoubleSignMonitor() {
 	}
 }
 
-=======
->>>>>>> f4d53133f (consensus, cmd, core, eth: remove support for non-merge mode of operation (#29169))
 // skipBlock returns 'true', if the block being imported can be skipped over, meaning
 // that the block does not need to be processed but can be considered already fully 'done'.
 func (bc *BlockChain) skipBlock(err error, it *insertIterator) bool {

core/blockchain_insert.go

@@ -187,3 +187,8 @@ func (it *insertIterator) first() *types.Block {
 func (it *insertIterator) remaining() int {
 	return len(it.chain) - it.index
 }
+
+// processed returns the number of processed blocks.
+func (it *insertIterator) processed() int {
+	return it.index + 1
+}