Merge pull request #3051 from OffchainLabs/reorg_finality_data

Handle reorg issues when setting finality data in Execution

Author: eljobe

arbnode/consensus_execution_syncer.go

@@ -39,19 +39,22 @@ type ConsensusExecutionSyncer struct {
 	inboxReader    *InboxReader
 	execClient     execution.ExecutionClient
 	blockValidator *staker.BlockValidator
+	txStreamer     *TransactionStreamer
 }
 
 func NewConsensusExecutionSyncer(
 	config func() *ConsensusExecutionSyncerConfig,
 	inboxReader *InboxReader,
 	execClient execution.ExecutionClient,
 	blockValidator *staker.BlockValidator,
+	txStreamer *TransactionStreamer,
 ) *ConsensusExecutionSyncer {
 	return &ConsensusExecutionSyncer{
 		config:         config,
 		inboxReader:    inboxReader,
 		execClient:     execClient,
 		blockValidator: blockValidator,
+		txStreamer:     txStreamer,
 	}
 }
 
@@ -60,37 +63,61 @@ func (c *ConsensusExecutionSyncer) Start(ctx_in context.Context) {
 	c.CallIteratively(c.pushFinalityDataFromConsensusToExecution)
 }
 
+func (c *ConsensusExecutionSyncer) getFinalityData(
+	ctx context.Context,
+	msgCount arbutil.MessageIndex,
+	errMsgCount error,
+	scenario string,
+) (*arbutil.FinalityData, error) {
+	if errors.Is(errMsgCount, headerreader.ErrBlockNumberNotSupported) {
+		log.Debug("Finality not supported, not pushing finality data to execution")
+		return nil, errMsgCount
+	} else if errMsgCount != nil {
+		log.Error("Error getting finality msg count", "scenario", scenario, "err", errMsgCount)
+		return nil, errMsgCount
+	}
+
+	if msgCount == 0 {
+		return nil, nil
+	}
+	msgIdx := msgCount - 1
+	msgResult, err := c.txStreamer.ResultAtMessageIndex(msgIdx)
+	if err != nil {
+		log.Error("Error getting message result", "msgIdx", msgIdx, "err", err)
+		return nil, err
+	}
+
+	finalityData := &arbutil.FinalityData{
+		MsgIdx:    msgIdx,
+		BlockHash: msgResult.BlockHash,
+	}
+	return finalityData, nil
+}
+
 func (c *ConsensusExecutionSyncer) pushFinalityDataFromConsensusToExecution(ctx context.Context) time.Duration {
 	safeMsgCount, err := c.inboxReader.GetSafeMsgCount(ctx)
-	if errors.Is(err, headerreader.ErrBlockNumberNotSupported) {
-		log.Info("Finality not supported, not pushing finality data to execution")
-		return c.config().SyncInterval
-	} else if err != nil {
-		log.Error("Error getting safe message count", "err", err)
+	safeFinalityData, err := c.getFinalityData(ctx, safeMsgCount, err, "safe")
+	if err != nil {
 		return c.config().SyncInterval
 	}
 
 	finalizedMsgCount, err := c.inboxReader.GetFinalizedMsgCount(ctx)
-	if errors.Is(err, headerreader.ErrBlockNumberNotSupported) {
-		log.Info("Finality not supported, not pushing finality data to execution")
-		return c.config().SyncInterval
-	} else if err != nil {
-		log.Error("Error getting finalized message count", "err", err)
+	finalizedFinalityData, err := c.getFinalityData(ctx, finalizedMsgCount, err, "finalized")
+	if err != nil {
 		return c.config().SyncInterval
 	}
 
+	var validatedFinalityData *arbutil.FinalityData
 	var validatedMsgCount arbutil.MessageIndex
 	if c.blockValidator != nil {
 		validatedMsgCount = c.blockValidator.GetValidated()
+		validatedFinalityData, err = c.getFinalityData(ctx, validatedMsgCount, nil, "validated")
+		if err != nil {
+			return c.config().SyncInterval
+		}
 	}
 
-	finalityData := &arbutil.FinalityData{
-		SafeMsgCount:      safeMsgCount,
-		FinalizedMsgCount: finalizedMsgCount,
-		ValidatedMsgCount: &validatedMsgCount,
-	}
-
-	_, err = c.execClient.SetFinalityData(ctx, finalityData).Await(ctx)
+	_, err = c.execClient.SetFinalityData(ctx, safeFinalityData, finalizedFinalityData, validatedFinalityData).Await(ctx)
 	if err != nil {
 		log.Error("Error pushing finality data from consensus to execution", "err", err)
 	} else {

arbnode/inbox_test.go

@@ -71,7 +71,12 @@ func (w *execClientWrapper) FullSyncProgressMap(ctx context.Context) map[string]
 	w.t.Error("not supported")
 	return nil
 }
-func (w *execClientWrapper) SetFinalityData(ctx context.Context, finalityData *arbutil.FinalityData) containers.PromiseInterface[struct{}] {
+func (w *execClientWrapper) SetFinalityData(
+	ctx context.Context,
+	safeFinalityData *arbutil.FinalityData,
+	finalizedFinalityData *arbutil.FinalityData,
+	validatedFinalityData *arbutil.FinalityData,
+) containers.PromiseInterface[struct{}] {
 	return containers.NewReadyPromise(struct{}{}, nil)
 }
 

arbnode/node.go

@@ -1088,7 +1088,7 @@ func createNodeImpl(
 	consensusExecutionSyncerConfigFetcher := func() *ConsensusExecutionSyncerConfig {
 		return &configFetcher.Get().ConsensusExecutionSyncer
 	}
-	consensusExecutionSyncer := NewConsensusExecutionSyncer(consensusExecutionSyncerConfigFetcher, inboxReader, executionClient, blockValidator)
+	consensusExecutionSyncer := NewConsensusExecutionSyncer(consensusExecutionSyncerConfigFetcher, inboxReader, executionClient, blockValidator, txStreamer)
 
 	return &Node{
 		ArbDB:                    arbDb,
@@ -1407,6 +1407,9 @@ func (n *Node) Start(ctx context.Context) error {
 }
 
 func (n *Node) StopAndWait() {
+	if n.ConsensusExecutionSyncer != nil {
+		n.ConsensusExecutionSyncer.StopAndWait()
+	}
 	if n.MaintenanceRunner != nil && n.MaintenanceRunner.Started() {
 		n.MaintenanceRunner.StopAndWait()
 	}

arbutil/finality_data.go

@@ -3,8 +3,9 @@
 
 package arbutil
 
+import "github.com/ethereum/go-ethereum/common"
+
 type FinalityData struct {
-	FinalizedMsgCount MessageIndex
-	SafeMsgCount      MessageIndex
-	ValidatedMsgCount *MessageIndex
+	MsgIdx    MessageIndex
+	BlockHash common.Hash
 }

execution/gethexec/executionengine.go

@@ -299,6 +299,17 @@ func (s *ExecutionEngine) Reorg(msgIdxOfFirstMsgToAdd arbutil.MessageIndex, newM
 		return nil, nil
 	}
 
+	currentSafeBlock := s.bc.CurrentSafeBlock()
+	if currentSafeBlock != nil && lastBlockToKeep.Number().Cmp(currentSafeBlock.Number) < 0 {
+		log.Warn("reorg target block is below safe block", "lastBlockNumToKeep", lastBlockNumToKeep, "currentSafeBlock", currentSafeBlock.Number)
+		s.bc.SetSafe(nil)
+	}
+	currentFinalBlock := s.bc.CurrentFinalBlock()
+	if currentFinalBlock != nil && lastBlockToKeep.Number().Cmp(currentFinalBlock.Number) < 0 {
+		log.Warn("reorg target block is below final block", "lastBlockNumToKeep", lastBlockNumToKeep, "currentFinalBlock", currentFinalBlock.Number)
+		s.bc.SetFinalized(nil)
+	}
+
 	tag := s.bc.StateCache().WasmCacheTag()
 	// reorg Rust-side VM state
 	C.stylus_reorg_vm(C.uint64_t(lastBlockNumToKeep), C.uint32_t(tag))
@@ -1090,23 +1101,3 @@ func (s *ExecutionEngine) Maintenance(capLimit uint64) error {
 	defer s.createBlocksMutex.Unlock()
 	return s.bc.FlushTrieDB(common.StorageSize(capLimit))
 }
-
-func (s *ExecutionEngine) SetFinalized(finalizedBlockNumber uint64) error {
-	block := s.bc.GetBlockByNumber(finalizedBlockNumber)
-	if block == nil {
-		return errors.New("unable to get block by number")
-	}
-
-	s.bc.SetFinalized(block.Header())
-	return nil
-}
-
-func (s *ExecutionEngine) SetSafe(safeBlockNumber uint64) error {
-	block := s.bc.GetBlockByNumber(safeBlockNumber)
-	if block == nil {
-		return errors.New("unable to get block by number")
-	}
-
-	s.bc.SetSafe(block.Header())
-	return nil
-}

execution/gethexec/node.go

@@ -509,8 +509,13 @@ func (n *ExecutionNode) FullSyncProgressMap(ctx context.Context) map[string]inte
 	return n.SyncMonitor.FullSyncProgressMap(ctx)
 }
 
-func (n *ExecutionNode) SetFinalityData(ctx context.Context, finalityData *arbutil.FinalityData) containers.PromiseInterface[struct{}] {
-	err := n.SyncMonitor.SetFinalityData(ctx, finalityData)
+func (n *ExecutionNode) SetFinalityData(
+	ctx context.Context,
+	safeFinalityData *arbutil.FinalityData,
+	finalizedFinalityData *arbutil.FinalityData,
+	validatedFinalityData *arbutil.FinalityData,
+) containers.PromiseInterface[struct{}] {
+	err := n.SyncMonitor.SetFinalityData(ctx, safeFinalityData, finalizedFinalityData, validatedFinalityData)
 	return containers.NewReadyPromise(struct{}{}, err)
 }
 

execution/gethexec/sync_monitor.go

@@ -3,6 +3,7 @@ package gethexec
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	flag "github.com/spf13/pflag"
 
@@ -123,47 +124,73 @@ func (s *SyncMonitor) BlockMetadataByNumber(ctx context.Context, blockNum uint64
 	return nil, nil
 }
 
-func (s *SyncMonitor) getFinalityBlock(
+func (s *SyncMonitor) getFinalityBlockHeader(
 	waitForBlockValidator bool,
-	validatedMsgCount *arbutil.MessageIndex,
-	finalityMsgCount arbutil.MessageIndex,
-) (*types.Block, error) {
+	validatedFinalityData *arbutil.FinalityData,
+	finalityFinalityData *arbutil.FinalityData,
+) (*types.Header, error) {
+	if finalityFinalityData == nil {
+		return nil, nil
+	}
+
+	finalityMsgIdx := finalityFinalityData.MsgIdx
+	finalityBlockHash := finalityFinalityData.BlockHash
 	if waitForBlockValidator {
-		if validatedMsgCount == nil {
+		if validatedFinalityData == nil {
 			return nil, errors.New("block validator not set")
 		}
-		if finalityMsgCount > *validatedMsgCount {
-			finalityMsgCount = *validatedMsgCount
+		if finalityFinalityData.MsgIdx > validatedFinalityData.MsgIdx {
+			finalityMsgIdx = validatedFinalityData.MsgIdx
+			finalityBlockHash = validatedFinalityData.BlockHash
 		}
 	}
-	finalityBlockNumber := s.exec.MessageIndexToBlockNumber(finalityMsgCount - 1)
+
+	finalityBlockNumber := s.exec.MessageIndexToBlockNumber(finalityMsgIdx)
 	finalityBlock := s.exec.bc.GetBlockByNumber(finalityBlockNumber)
 	if finalityBlock == nil {
-		return nil, errors.New("unable to get block by number")
+		log.Debug("Finality block not found", "blockNumber", finalityBlockNumber)
+		return nil, nil
+	}
+	if finalityBlock.Hash() != finalityBlockHash {
+		errorMsg := fmt.Sprintf(
+			"finality block hash mismatch, blockNumber=%v, block hash provided by consensus=%v, block hash from execution=%v",
+			finalityBlockNumber,
+			finalityBlockHash,
+			finalityBlock.Hash(),
+		)
+		return nil, errors.New(errorMsg)
 	}
-	return finalityBlock, nil
+	return finalityBlock.Header(), nil
 }
 
-func (s *SyncMonitor) SetFinalityData(ctx context.Context, finalityData *arbutil.FinalityData) error {
-	finalizedBlock, err := s.getFinalityBlock(
+func (s *SyncMonitor) SetFinalityData(
+	ctx context.Context,
+	safeFinalityData *arbutil.FinalityData,
+	finalizedFinalityData *arbutil.FinalityData,
+	validatedFinalityData *arbutil.FinalityData,
+) error {
+	s.exec.createBlocksMutex.Lock()
+	defer s.exec.createBlocksMutex.Unlock()
+
+	finalizedBlockHeader, err := s.getFinalityBlockHeader(
 		s.config.FinalizedBlockWaitForBlockValidator,
-		finalityData.ValidatedMsgCount,
-		finalityData.FinalizedMsgCount,
+		validatedFinalityData,
+		finalizedFinalityData,
 	)
 	if err != nil {
 		return err
 	}
-	s.exec.bc.SetFinalized(finalizedBlock.Header())
+	s.exec.bc.SetFinalized(finalizedBlockHeader)
 
-	safeBlock, err := s.getFinalityBlock(
+	safeBlockHeader, err := s.getFinalityBlockHeader(
 		s.config.SafeBlockWaitForBlockValidator,
-		finalityData.ValidatedMsgCount,
-		finalityData.SafeMsgCount,
+		validatedFinalityData,
+		safeFinalityData,
 	)
 	if err != nil {
 		return err
 	}
-	s.exec.bc.SetSafe(safeBlock.Header())
+	s.exec.bc.SetSafe(safeBlockHeader)
 
 	return nil
 }

execution/interface.go

@@ -40,7 +40,7 @@ type ExecutionClient interface {
 	ResultAtMessageIndex(msgIdx arbutil.MessageIndex) containers.PromiseInterface[*MessageResult]
 	MessageIndexToBlockNumber(messageNum arbutil.MessageIndex) containers.PromiseInterface[uint64]
 	BlockNumberToMessageIndex(blockNum uint64) containers.PromiseInterface[arbutil.MessageIndex]
-	SetFinalityData(ctx context.Context, finalityData *arbutil.FinalityData) containers.PromiseInterface[struct{}]
+	SetFinalityData(ctx context.Context, safeFinalityData *arbutil.FinalityData, finalizedFinalityData *arbutil.FinalityData, validatedFinalityData *arbutil.FinalityData) containers.PromiseInterface[struct{}]
 	MarkFeedStart(to arbutil.MessageIndex) containers.PromiseInterface[struct{}]
 
 	Maintenance() containers.PromiseInterface[struct{}]