consensus_execution_syncer.go

// Copyright 2021-2026, Offchain Labs, Inc.
// For license information, see https://github.com/OffchainLabs/nitro/blob/master/LICENSE.md

package arbnode

import (
	"context"
	"errors"
	"time"

	"github.com/spf13/pflag"

	"github.com/ethereum/go-ethereum/log"

	"github.com/offchainlabs/nitro/arbutil"
	"github.com/offchainlabs/nitro/execution"
	"github.com/offchainlabs/nitro/staker"
	"github.com/offchainlabs/nitro/util"
	"github.com/offchainlabs/nitro/util/headerreader"
	"github.com/offchainlabs/nitro/util/stopwaiter"
)

type ConsensusExecutionSyncerConfig struct {
	SyncInterval time.Duration `koanf:"sync-interval"`
}

var DefaultConsensusExecutionSyncerConfig = ConsensusExecutionSyncerConfig{
	SyncInterval: 300 * time.Millisecond,
}

var TestConsensusExecutionSyncerConfig = ConsensusExecutionSyncerConfig{
	SyncInterval: TestSyncMonitorConfig.MsgLag / 2,
}

// We don't define a Test config. For most tests we want the Syncer to behave
// the same as in production.

func ConsensusExecutionSyncerConfigAddOptions(prefix string, f *pflag.FlagSet) {
	f.Duration(prefix+".sync-interval", DefaultConsensusExecutionSyncerConfig.SyncInterval, "Interval in which finality and sync data is pushed from consensus to execution")
}

type MessageCountFetcher interface {
	GetSafeMsgCount(ctx context.Context) (arbutil.MessageIndex, error)
	GetFinalizedMsgCount(ctx context.Context) (arbutil.MessageIndex, error)
	SupportsPushingFinalityData() bool
}

// lint:require-exhaustive-initialization
type ConsensusExecutionSyncer struct {
	stopwaiter.StopWaiter

	config func() *ConsensusExecutionSyncerConfig

	msgCountFetcher      MessageCountFetcher
	execClient           execution.ExecutionClient
	blockValidator       *staker.BlockValidator
	txStreamer           *TransactionStreamer
	syncMonitor          *SyncMonitor
	msgCountErrorHandler *util.EphemeralErrorHandler
}

func NewConsensusExecutionSyncer(
	config func() *ConsensusExecutionSyncerConfig,
	msgCountFetcher MessageCountFetcher,
	execClient execution.ExecutionClient,
	blockValidator *staker.BlockValidator,
	txStreamer *TransactionStreamer,
	syncMonitor *SyncMonitor,
) *ConsensusExecutionSyncer {
	return &ConsensusExecutionSyncer{
		StopWaiter:      stopwaiter.StopWaiter{},
		config:          config,
		msgCountFetcher: msgCountFetcher,
		execClient:      execClient,
		blockValidator:  blockValidator,
		txStreamer:      txStreamer,
		syncMonitor:     syncMonitor,
		// For the first 2 minutes, log msg count error as WARN, then as ERROR.
		msgCountErrorHandler: util.NewEphemeralErrorHandler(2*time.Minute, "", 0),
	}
}

func (c *ConsensusExecutionSyncer) Start(ctx_in context.Context) {
	c.StopWaiter.Start(ctx_in, c)
	if c.msgCountFetcher != nil && c.msgCountFetcher.SupportsPushingFinalityData() {
		c.CallIteratively(c.pushFinalityDataFromConsensusToExecution)
	}
	c.CallIteratively(c.pushConsensusSyncDataToExecution)
}

func (c *ConsensusExecutionSyncer) getFinalityData(
	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 {
		c.msgCountErrorHandler.LogLevel(errMsgCount, log.Error)("Error getting finality msg count", "scenario", scenario, "err", errMsgCount)
		return nil, errMsgCount
	}
	c.msgCountErrorHandler.Reset()

	if msgCount == 0 {
		return nil, nil
	}
	msgIdx := msgCount - 1
	msgResult, err := c.txStreamer.ResultAtMessageIndex(msgIdx)
	if errors.Is(err, execution.ErrResultNotFound) {
		log.Debug("Message result not found, node out of sync", "msgIdx", msgIdx, "err", err)
		return nil, nil
	} else 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.msgCountFetcher.GetSafeMsgCount(ctx)
	if err != nil {
		return c.config().SyncInterval
	}
	safeFinalityData, err := c.getFinalityData(safeMsgCount, err, "safe")
	if err != nil {
		return c.config().SyncInterval
	}

	finalizedMsgCount, err := c.msgCountFetcher.GetFinalizedMsgCount(ctx)
	if err != nil {
		return c.config().SyncInterval
	}
	finalizedFinalityData, err := c.getFinalityData(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(validatedMsgCount, nil, "validated")
		if err != nil {
			return c.config().SyncInterval
		}
	}

	_, err = c.execClient.SetFinalityData(safeFinalityData, finalizedFinalityData, validatedFinalityData).Await(ctx)
	if err != nil {
		log.Error("Error pushing finality data from consensus to execution", "err", err)
	} else {
		finalityMsgCount := func(fd *arbutil.FinalityData) arbutil.MessageIndex {
			if fd != nil {
				return fd.MsgIdx + 1
			}
			return 0
		}
		log.Debug("Pushed finality data from consensus to execution",
			"safeMsgCount", finalityMsgCount(safeFinalityData),
			"finalizedMsgCount", finalityMsgCount(finalizedFinalityData),
			"validatedMsgCount", finalityMsgCount(validatedFinalityData),
		)
	}

	return c.config().SyncInterval
}

func (c *ConsensusExecutionSyncer) pushConsensusSyncDataToExecution(ctx context.Context) time.Duration {
	synced := c.syncMonitor.Synced()

	maxMessageCount, err := c.syncMonitor.GetMaxMessageCount()
	if err != nil {
		log.Error("Error getting max message count", "err", err)
		return c.config().SyncInterval
	}

	var syncProgressMap map[string]interface{}
	if !synced {
		// Only populate the full progress map when not synced (for debugging)
		syncProgressMap = c.syncMonitor.FullSyncProgressMap()
	}

	syncData := &execution.ConsensusSyncData{
		Synced:          synced,
		MaxMessageCount: maxMessageCount,
		SyncProgressMap: syncProgressMap,
		UpdatedAt:       time.Now(),
	}

	_, err = c.execClient.SetConsensusSyncData(syncData).Await(ctx)
	if err != nil {
		log.Error("Error pushing sync data from consensus to execution", "err", err)
	} else {
		log.Debug("Pushed sync data from consensus to execution",
			"synced", syncData.Synced,
			"maxMessageCount", syncData.MaxMessageCount,
			"updatedAt", syncData.UpdatedAt,
			"hasProgressMap", syncData.SyncProgressMap != nil,
		)
	}

	return c.config().SyncInterval
}