feat: refactor miner/dBFT to properly support post-merge behaviour

Author: qianhh

consensus/consensus.go

@@ -118,3 +118,11 @@ type Engine interface {
 	// Close terminates any background threads maintained by the consensus engine.
 	Close() error
 }
+
+// PoS consensus engine.
+type PoS interface {
+	Engine
+
+	// GetFinalizedHeader retrieves the finalized header for a given header.
+	GetFinalizedHeader(chain ChainHeaderReader, header *types.Header) *types.Header
+}

consensus/dbft/dbft.go

@@ -2948,3 +2948,15 @@ func unpackContractExecutionResult(res interface{}, result *core.ExecutionResult
 	}
 	return contractAbi.UnpackIntoInterface(&res, method, result.Return())
 }
+
+// GetFinalizedHeader returns highest finalized block header.
+func (c *DBFT) GetFinalizedHeader(chain consensus.ChainHeaderReader, header *types.Header) *types.Header {
+	if chain == nil || header == nil {
+		return nil
+	}
+	if !chain.Config().IsNeoXDKG(header.Number) || header.Number.Uint64() < 1 {
+		return chain.GetHeaderByNumber(0)
+	}
+
+	return chain.GetHeaderByNumber(header.Number.Uint64() - 1)
+}

core/blockchain.go

@@ -225,14 +225,15 @@ type BlockChain struct {
 	statedb       *state.CachingDB                 // State database to reuse between imports (contains state cache)
 	txIndexer     *txIndexer                       // Transaction indexer, might be nil if not enabled
 
-	hc            *HeaderChain
-	rmLogsFeed    event.Feed
-	chainFeed     event.Feed
-	chainHeadFeed event.Feed
-	logsFeed      event.Feed
-	blockProcFeed event.Feed
-	scope         event.SubscriptionScope
-	genesisBlock  *types.Block
+	hc                  *HeaderChain
+	rmLogsFeed          event.Feed
+	chainFeed           event.Feed
+	chainHeadFeed       event.Feed
+	logsFeed            event.Feed
+	blockProcFeed       event.Feed
+	finalizedHeaderFeed event.Feed
+	scope               event.SubscriptionScope
+	genesisBlock        *types.Block
 
 	// This mutex synchronizes chain write operations.
 	// Readers don't need to take it, they can just read the database.
@@ -497,6 +498,17 @@ func (bc *BlockChain) empty() bool {
 	return true
 }
 
+// GetFinalizedNumber returns the highest finalized number before the specific block.
+func (bc *BlockChain) GetFinalizedNumber(header *types.Header) uint64 {
+	if p, ok := bc.engine.(consensus.PoS); ok {
+		if finalizedHeader := p.GetFinalizedHeader(bc, header); finalizedHeader != nil {
+			return finalizedHeader.Number.Uint64()
+		}
+	}
+
+	return 0
+}
+
 // loadLastState loads the last known chain state from the database. This method
 // assumes that the chain manager mutex is held.
 func (bc *BlockChain) loadLastState() error {
@@ -551,8 +563,7 @@ func (bc *BlockChain) loadLastState() error {
 	}
 	// Issue a status log for the user
 	var (
-		currentSnapBlock  = bc.CurrentSnapBlock()
-		currentFinalBlock = bc.CurrentFinalBlock()
+		currentSnapBlock = bc.CurrentSnapBlock()
 
 		headerTd = bc.GetTd(headHeader.Hash(), headHeader.Number.Uint64())
 		blockTd  = bc.GetTd(headBlock.Hash(), headBlock.NumberU64())
@@ -565,9 +576,14 @@ func (bc *BlockChain) loadLastState() error {
 		snapTd := bc.GetTd(currentSnapBlock.Hash(), currentSnapBlock.Number.Uint64())
 		log.Info("Loaded most recent local snap block", "number", currentSnapBlock.Number, "hash", currentSnapBlock.Hash(), "td", snapTd, "age", common.PrettyAge(time.Unix(int64(currentSnapBlock.Time), 0)))
 	}
-	if currentFinalBlock != nil {
-		finalTd := bc.GetTd(currentFinalBlock.Hash(), currentFinalBlock.Number.Uint64())
-		log.Info("Loaded most recent local finalized block", "number", currentFinalBlock.Number, "hash", currentFinalBlock.Hash(), "td", finalTd, "age", common.PrettyAge(time.Unix(int64(currentFinalBlock.Time), 0)))
+	if p, ok := bc.engine.(consensus.PoS); ok {
+		if currentFinalizedHeader := p.GetFinalizedHeader(bc, headHeader); currentFinalizedHeader != nil {
+			bc.currentFinalBlock.Store(currentFinalizedHeader)
+			if currentFinalizedBlock := bc.GetBlockByHash(currentFinalizedHeader.Hash()); currentFinalizedBlock != nil {
+				finalTd := bc.GetTd(currentFinalizedBlock.Hash(), currentFinalizedBlock.NumberU64())
+				log.Info("Loaded most recent local finalized block", "number", currentFinalizedBlock.Number(), "hash", currentFinalizedBlock.Hash(), "root", currentFinalizedBlock.Root(), "td", finalTd, "age", common.PrettyAge(time.Unix(int64(currentFinalizedBlock.Time()), 0)))
+			}
+		}
 	}
 	if pivot := rawdb.ReadLastPivotNumber(bc.db); pivot != nil {
 		log.Info("Loaded last snap-sync pivot marker", "number", *pivot)
@@ -1615,13 +1631,24 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 		if len(logs) > 0 {
 			bc.logsFeed.Send(logs)
 		}
+
+		var finalizedHeader *types.Header
+		if p, ok := bc.Engine().(consensus.PoS); ok {
+			if finalizedHeader = p.GetFinalizedHeader(bc, block.Header()); finalizedHeader != nil {
+				bc.SetFinalized(finalizedHeader)
+			}
+		}
+
 		// In theory, we should fire a ChainHeadEvent when we inject
 		// a canonical block, but sometimes we can insert a batch of
 		// canonical blocks. Avoid firing too many ChainHeadEvents,
 		// we will fire an accumulated ChainHeadEvent and disable fire
 		// event here.
 		if emitHeadEvent {
 			bc.chainHeadFeed.Send(ChainHeadEvent{Header: block.Header()})
+			if finalizedHeader != nil {
+				bc.finalizedHeaderFeed.Send(FinalizedHeaderEvent{finalizedHeader})
+			}
 		}
 	}
 	return status, nil
@@ -1708,6 +1735,11 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness
 	defer func() {
 		if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() {
 			bc.chainHeadFeed.Send(ChainHeadEvent{Header: lastCanon.Header()})
+			if p, ok := bc.Engine().(consensus.PoS); ok {
+				if finalizedHeader := p.GetFinalizedHeader(bc, lastCanon.Header()); finalizedHeader != nil {
+					bc.finalizedHeaderFeed.Send(FinalizedHeaderEvent{finalizedHeader})
+				}
+			}
 		}
 	}()
 	// Start the parallel header verifier

core/blockchain_reader.go

@@ -54,13 +54,29 @@ func (bc *BlockChain) CurrentSnapBlock() *types.Header {
 // CurrentFinalBlock retrieves the current finalized block of the canonical
 // chain. The block is retrieved from the blockchain's internal cache.
 func (bc *BlockChain) CurrentFinalBlock() *types.Header {
-	return bc.currentFinalBlock.Load()
+	if p, ok := bc.engine.(consensus.PoS); ok {
+		currentHeader := bc.CurrentHeader()
+		if currentHeader == nil {
+			return nil
+		}
+		return p.GetFinalizedHeader(bc, currentHeader)
+	}
+
+	return nil
 }
 
 // CurrentSafeBlock retrieves the current safe block of the canonical
 // chain. The block is retrieved from the blockchain's internal cache.
 func (bc *BlockChain) CurrentSafeBlock() *types.Header {
-	return bc.currentSafeBlock.Load()
+	if p, ok := bc.engine.(consensus.PoS); ok {
+		currentHeader := bc.CurrentHeader()
+		if currentHeader == nil {
+			return nil
+		}
+		return p.GetFinalizedHeader(bc, currentHeader)
+	}
+
+	return nil
 }
 
 // HasHeader checks if a block header is present in the database or not, caching
@@ -440,3 +456,8 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
 func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
 	return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
 }
+
+// SubscribeFinalizedHeaderEvent registers a subscription of FinalizedHeaderEvent.
+func (bc *BlockChain) SubscribeFinalizedHeaderEvent(ch chan<- FinalizedHeaderEvent) event.Subscription {
+	return bc.scope.Track(bc.finalizedHeaderFeed.Subscribe(ch))
+}

core/events.go

@@ -32,6 +32,9 @@ type NewMinedBlockEvent struct{ Block *types.Block }
 // RemovedLogsEvent is posted when a reorg happens
 type RemovedLogsEvent struct{ Logs []*types.Log }
 
+// FinalizedHeaderEvent is posted when a finalized header is reached.
+type FinalizedHeaderEvent struct{ Header *types.Header }
+
 type ChainEvent struct {
 	Header *types.Header
 }

core/headerchain.go

@@ -99,6 +99,17 @@ func NewHeaderChain(chainDb ethdb.Database, config *params.ChainConfig, engine c
 	return hc, nil
 }
 
+// GetFinalizedNumber returns the highest finalized number before the specific block.
+func (hc *HeaderChain) GetFinalizedNumber(header *types.Header) uint64 {
+	if p, ok := hc.engine.(consensus.PoS); ok {
+		if finalizedHeader := p.GetFinalizedHeader(hc, header); finalizedHeader != nil {
+			return finalizedHeader.Number.Uint64()
+		}
+	}
+
+	return 0
+}
+
 // GetBlockNumber retrieves the block number belonging to the given hash
 // from the cache or database
 func (hc *HeaderChain) GetBlockNumber(hash common.Hash) *uint64 {

eth/api_backend.go

@@ -277,6 +277,10 @@ func (b *EthAPIBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) e
 	return b.eth.BlockChain().SubscribeChainHeadEvent(ch)
 }
 
+func (b *EthAPIBackend) SubscribeFinalizedHeaderEvent(ch chan<- core.FinalizedHeaderEvent) event.Subscription {
+	return b.eth.BlockChain().SubscribeFinalizedHeaderEvent(ch)
+}
+
 func (b *EthAPIBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
 	return b.eth.BlockChain().SubscribeLogsEvent(ch)
 }

eth/ethconfig/config.go

@@ -187,19 +187,19 @@ 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, statisticsCfg dbft.StatisticsConfig) (consensus.Engine, error) {
+	if config.DBFT != nil {
+		bft, err := dbft.New(config, db, statisticsCfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create dBFT engine: %w", err)
+		}
+		return bft, nil
+	}
 	if config.TerminalTotalDifficulty == nil {
 		return nil, fmt.Errorf("only PoS networks are supported, please transition old ones with Geth v1.13.x")
 	}
 	// If proof-of-authority is requested, set it up
 	if config.Clique != nil {
 		return beacon.New(clique.New(config.Clique, db)), nil
 	}
-	if config.DBFT != nil {
-		bft, err := dbft.New(config, db, statisticsCfg)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create dBFT engine: %w", err)
-		}
-		return beacon.New(bft), nil
-	}
 	return beacon.New(ethash.NewFaker()), nil
 }

eth/filters/api.go

@@ -269,6 +269,65 @@ func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
 	return rpcSub, nil
 }
 
+// NewFinalizedHeaderFilter creates a filter that fetches finalized headers that are reached.
+func (api *FilterAPI) NewFinalizedHeaderFilter() rpc.ID {
+	var (
+		headers   = make(chan *types.Header)
+		headerSub = api.events.SubscribeNewFinalizedHeaders(headers)
+	)
+
+	api.filtersMu.Lock()
+	api.filters[headerSub.ID] = &filter{typ: FinalizedHeadersSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: headerSub}
+	api.filtersMu.Unlock()
+
+	go func() {
+		for {
+			select {
+			case h := <-headers:
+				api.filtersMu.Lock()
+				if f, found := api.filters[headerSub.ID]; found {
+					f.hashes = append(f.hashes, h.Hash())
+				}
+				api.filtersMu.Unlock()
+			case <-headerSub.Err():
+				api.filtersMu.Lock()
+				delete(api.filters, headerSub.ID)
+				api.filtersMu.Unlock()
+				return
+			}
+		}
+	}()
+
+	return headerSub.ID
+}
+
+// NewFinalizedHeaders send a notification each time a new finalized header is reached.
+func (api *FilterAPI) NewFinalizedHeaders(ctx context.Context) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
+	}
+
+	rpcSub := notifier.CreateSubscription()
+
+	go func() {
+		headers := make(chan *types.Header)
+		headersSub := api.events.SubscribeNewFinalizedHeaders(headers)
+		defer headersSub.Unsubscribe()
+
+		for {
+			select {
+			case h := <-headers:
+				notifier.Notify(rpcSub.ID, h)
+			case <-rpcSub.Err():
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
+
 // Logs creates a subscription that fires for all new log that match the given filter criteria.
 func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)