Merge pull request #7066 from ethereum-optimism/indexer.conf.depth

feat(indexer): configurable confirmation depth when traversing headers

indexer/config/config.go

@@ -2,7 +2,6 @@ package config
 
 import (
 	"fmt"
-	"math/big"
 	"os"
 	"reflect"
 
@@ -66,10 +65,15 @@ func (c *L1Contracts) AsSlice() ([]common.Address, error) {
 type ChainConfig struct {
 	// Configure known chains with the l2 chain id
 	Preset int
+
 	L1Contracts      L1Contracts `toml:"l1-contracts"`
-	// L1StartingHeight is the block height to start indexing from
 	L1StartingHeight uint        `toml:"l1-starting-height"`
 
+	// These configuration options will be removed once
+	// native reorg handling is implemented
+	L1ConfirmationDepth uint `toml:"l1-confirmation-depth"`
+	L2ConfirmationDepth uint `toml:"l2-confirmation-depth"`
+
 	L1PollingInterval uint `toml:"l1-polling-interval"`
 	L2PollingInterval uint `toml:"l2-polling-interval"`
 
@@ -77,11 +81,6 @@ type ChainConfig struct {
 	L2HeaderBufferSize uint `toml:"l2-header-buffer-size"`
 }
 
-// L1StartHeight returns the block height to start indexing from
-func (cc *ChainConfig) L1StartHeight() *big.Int {
-	return big.NewInt(int64(cc.L1StartingHeight))
-}
-
 // RPCsConfig configures the RPC urls
 type RPCsConfig struct {
 	L1RPC string `toml:"l1-rpc"`

indexer/e2e_tests/setup.go

@@ -71,8 +71,10 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 			L2RPC: opSys.EthInstances["sequencer"].HTTPEndpoint(),
 		},
 		Chain: config.ChainConfig{
-			L1PollingInterval: 1000,
-			L2PollingInterval: 1000,
+			L1PollingInterval:   uint(opCfg.DeployConfig.L1BlockTime) * 1000,
+			L1ConfirmationDepth: 0,
+			L2PollingInterval:   uint(opCfg.DeployConfig.L2BlockTime) * 1000,
+			L2ConfirmationDepth: 0,
 			L1Contracts: config.L1Contracts{
 				OptimismPortalProxy:         opCfg.L1Deployments.OptimismPortalProxy,
 				L2OutputOracleProxy:         opCfg.L1Deployments.L2OutputOracleProxy,

indexer/etl/etl.go

@@ -16,7 +16,9 @@ import (
 type Config struct {
 	LoopIntervalMsec uint
 	HeaderBufferSize uint
+
 	StartHeight       *big.Int
+	ConfirmationDepth *big.Int
 }
 
 type ETL struct {

indexer/etl/l1_etl.go

@@ -65,7 +65,7 @@ func NewL1ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, cli
 
 		log:             log,
 		metrics:         metrics,
-		headerTraversal: node.NewHeaderTraversal(client, fromHeader),
+		headerTraversal: node.NewHeaderTraversal(client, fromHeader, cfg.ConfirmationDepth),
 		ethClient:       client,
 		contracts:       cSlice,
 		etlBatches:      etlBatches,

indexer/etl/l2_etl.go

@@ -49,7 +49,7 @@ func NewL2ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, cli
 
 		log:             log,
 		metrics:         metrics,
-		headerTraversal: node.NewHeaderTraversal(client, fromHeader),
+		headerTraversal: node.NewHeaderTraversal(client, fromHeader, cfg.ConfirmationDepth),
 		ethClient:       client,
 		contracts:       l2Contracts,
 		etlBatches:      etlBatches,

indexer/indexer.go

@@ -3,6 +3,7 @@ package indexer
 import (
 	"context"
 	"fmt"
+	"math/big"
 	"runtime/debug"
 	"sync"
 
@@ -28,7 +29,6 @@ type Indexer struct {
 
 	L1ETL           *etl.L1ETL
 	L2ETL           *etl.L2ETL
-
 	BridgeProcessor *processors.BridgeProcessor
 }
 
@@ -41,7 +41,12 @@ func NewIndexer(logger log.Logger, db *database.DB, chainConfig config.ChainConf
 	if err != nil {
 		return nil, err
 	}
-	l1Cfg := etl.Config{LoopIntervalMsec: chainConfig.L1PollingInterval, HeaderBufferSize: chainConfig.L1HeaderBufferSize, StartHeight: chainConfig.L1StartHeight()}
+	l1Cfg := etl.Config{
+		LoopIntervalMsec:  chainConfig.L1PollingInterval,
+		HeaderBufferSize:  chainConfig.L1HeaderBufferSize,
+		ConfirmationDepth: big.NewInt(int64(chainConfig.L1ConfirmationDepth)),
+		StartHeight:       big.NewInt(int64(chainConfig.L1StartingHeight)),
+	}
 	l1Etl, err := etl.NewL1ETL(l1Cfg, logger, db, etl.NewMetrics(metricsRegistry, "l1"), l1EthClient, chainConfig.L1Contracts)
 	if err != nil {
 		return nil, err
@@ -52,7 +57,11 @@ func NewIndexer(logger log.Logger, db *database.DB, chainConfig config.ChainConf
 	if err != nil {
 		return nil, err
 	}
-	l2Cfg := etl.Config{LoopIntervalMsec: chainConfig.L2PollingInterval, HeaderBufferSize: chainConfig.L2HeaderBufferSize}
+	l2Cfg := etl.Config{
+		LoopIntervalMsec:  chainConfig.L2PollingInterval,
+		HeaderBufferSize:  chainConfig.L2HeaderBufferSize,
+		ConfirmationDepth: big.NewInt(int64(chainConfig.L2ConfirmationDepth)),
+	}
 	l2Etl, err := etl.NewL2ETL(l2Cfg, logger, db, etl.NewMetrics(metricsRegistry, "l2"), l2EthClient)
 	if err != nil {
 		return nil, err

indexer/indexer.toml

 # Chain configures l1 chain addresses
 # Can configure them manually or use a preset l2 ChainId for known chains including OP Mainnet, OP Goerli, Base, Base Goerli, Zora, and Zora goerli
 [chain]
+
+# OP Goerli
+preset = 420
+
+# L1 Config
 l1-polling-interval = 0
 l1-header-buffer-size = 0
+l1-confirmation-depth = 0
+l1-starting-height = 0
 
+# L2 Config
 l2-polling-interval = 0
 l2-header-buffer-size = 0
+l2-confirmation-depth = 0
 
-# OP Goerli
-preset = 420
-l1-starting-height = 0
 
 [rpcs]
 l1-rpc = "${INDEXER_RPC_URL_L1}"

indexer/node/client.go

@@ -25,8 +25,6 @@ const (
 )
 
 type EthClient interface {
-	FinalizedBlockHeight() (*big.Int, error)
-
 	BlockHeaderByNumber(*big.Int) (*types.Header, error)
 	BlockHeaderByHash(common.Hash) (*types.Header, error)
 	BlockHeadersByRange(*big.Int, *big.Int) ([]types.Header, error)
@@ -52,24 +50,6 @@ func DialEthClient(rpcUrl string, metrics Metricer) (EthClient, error) {
 	return client, nil
 }
 
-// FinalizedBlockHeight retrieves the latest block height in a finalized state
-func (c *client) FinalizedBlockHeight() (*big.Int, error) {
-	ctxwt, cancel := context.WithTimeout(context.Background(), defaultRequestTimeout)
-	defer cancel()
-
-	// **NOTE** Local devnet is having issues with the "finalized" block tag. Temp switch
-	// to "latest" to iterate faster locally but this needs to be updated
-	var header *types.Header
-	err := c.rpc.CallContext(ctxwt, &header, "eth_getBlockByNumber", "latest", false)
-	if err != nil {
-		return nil, err
-	} else if header == nil {
-		return nil, ethereum.NotFound
-	}
-
-	return header.Number, nil
-}
-
 // BlockHeaderByHash retrieves the block header attributed to the supplied hash
 func (c *client) BlockHeaderByHash(hash common.Hash) (*types.Header, error) {
 	ctxwt, cancel := context.WithTimeout(context.Background(), defaultRequestTimeout)

indexer/node/header_traversal.go

@@ -15,14 +15,15 @@ var (
 
 type HeaderTraversal struct {
 	ethClient EthClient
+
 	lastHeader             *types.Header
+	blockConfirmationDepth *big.Int
 }
 
 // NewHeaderTraversal instantiates a new instance of HeaderTraversal against the supplied rpc client.
-// The HeaderTraversal will start fetching blocks starting from the supplied header unless
-// nil, indicating genesis.
-func NewHeaderTraversal(ethClient EthClient, fromHeader *types.Header) *HeaderTraversal {
-	return &HeaderTraversal{ethClient: ethClient, lastHeader: fromHeader}
+// The HeaderTraversal will start fetching blocks starting from the supplied header unless nil, indicating genesis.
+func NewHeaderTraversal(ethClient EthClient, fromHeader *types.Header, confDepth *big.Int) *HeaderTraversal {
+	return &HeaderTraversal{ethClient: ethClient, lastHeader: fromHeader, blockConfirmationDepth: confDepth}
 }
 
 // LastHeader returns the last header that was fetched by the HeaderTraversal
@@ -34,13 +35,19 @@ func (f *HeaderTraversal) LastHeader() *types.Header {
 // NextFinalizedHeaders retrives the next set of headers that have been
 // marked as finalized by the connected client, bounded by the supplied size
 func (f *HeaderTraversal) NextFinalizedHeaders(maxSize uint64) ([]types.Header, error) {
-	finalizedBlockHeight, err := f.ethClient.FinalizedBlockHeight()
+	latestBlockHeader, err := f.ethClient.BlockHeaderByNumber(nil)
 	if err != nil {
-		return nil, fmt.Errorf("unable to query latest finalized height: %w", err)
+		return nil, fmt.Errorf("unable to query latest block: %w", err)
+	}
+
+	endHeight := new(big.Int).Sub(latestBlockHeader.Number, f.blockConfirmationDepth)
+	if endHeight.Sign() < 0 {
+		// No blocks with the provided confirmation depth available
+		return nil, nil
 	}
 
 	if f.lastHeader != nil {
-		cmp := f.lastHeader.Number.Cmp(finalizedBlockHeight)
+		cmp := f.lastHeader.Number.Cmp(endHeight)
 		if cmp == 0 {
 			return nil, nil
 		} else if cmp > 0 {
@@ -53,7 +60,7 @@ func (f *HeaderTraversal) NextFinalizedHeaders(maxSize uint64) ([]types.Header,
 		nextHeight = new(big.Int).Add(f.lastHeader.Number, bigOne)
 	}
 
-	endHeight := clampBigInt(nextHeight, finalizedBlockHeight, maxSize)
+	endHeight = clampBigInt(nextHeight, endHeight, maxSize)
 	headers, err := f.ethClient.BlockHeadersByRange(nextHeight, endHeight)
 	if err != nil {
 		return nil, fmt.Errorf("error querying blocks by range: %w", err)

indexer/node/header_traversal_test.go

@@ -37,10 +37,10 @@ func TestHeaderTraversalNextFinalizedHeadersNoOp(t *testing.T) {
 
 	// start from block 10 as the latest fetched block
 	lastHeader := &types.Header{Number: big.NewInt(10)}
-	headerTraversal := NewHeaderTraversal(client, lastHeader)
+	headerTraversal := NewHeaderTraversal(client, lastHeader, bigZero)
 
 	// no new headers when matched with head
-	client.On("FinalizedBlockHeight").Return(big.NewInt(10), nil)
+	client.On("BlockHeaderByNumber", (*big.Int)(nil)).Return(lastHeader, nil)
 	headers, err := headerTraversal.NextFinalizedHeaders(100)
 	require.NoError(t, err)
 	require.Empty(t, headers)
@@ -50,11 +50,11 @@ func TestHeaderTraversalNextFinalizedHeadersCursored(t *testing.T) {
 	client := new(MockEthClient)
 
 	// start from genesis
-	headerTraversal := NewHeaderTraversal(client, nil)
+	headerTraversal := NewHeaderTraversal(client, nil, bigZero)
 
 	// blocks [0..4]
 	headers := makeHeaders(5, nil)
-	client.On("FinalizedBlockHeight").Return(big.NewInt(4), nil).Times(1) // Times so that we can override next
+	client.On("BlockHeaderByNumber", (*big.Int)(nil)).Return(&headers[4], nil).Times(1) // Times so that we can override next
 	client.On("BlockHeadersByRange", mock.MatchedBy(BigIntMatcher(0)), mock.MatchedBy(BigIntMatcher(4))).Return(headers, nil)
 	headers, err := headerTraversal.NextFinalizedHeaders(5)
 	require.NoError(t, err)
@@ -62,7 +62,7 @@ func TestHeaderTraversalNextFinalizedHeadersCursored(t *testing.T) {
 
 	// blocks [5..9]
 	headers = makeHeaders(5, &headers[len(headers)-1])
-	client.On("FinalizedBlockHeight").Return(big.NewInt(9), nil)
+	client.On("BlockHeaderByNumber", (*big.Int)(nil)).Return(&headers[4], nil)
 	client.On("BlockHeadersByRange", mock.MatchedBy(BigIntMatcher(5)), mock.MatchedBy(BigIntMatcher(9))).Return(headers, nil)
 	headers, err = headerTraversal.NextFinalizedHeaders(5)
 	require.NoError(t, err)
@@ -73,10 +73,10 @@ func TestHeaderTraversalNextFinalizedHeadersMaxSize(t *testing.T) {
 	client := new(MockEthClient)
 
 	// start from genesis
-	headerTraversal := NewHeaderTraversal(client, nil)
+	headerTraversal := NewHeaderTraversal(client, nil, bigZero)
 
 	// 100 "available" headers
-	client.On("FinalizedBlockHeight").Return(big.NewInt(100), nil)
+	client.On("BlockHeaderByNumber", (*big.Int)(nil)).Return(&types.Header{Number: big.NewInt(100)}, nil)
 
 	// clamped by the supplied size
 	headers := makeHeaders(5, nil)
@@ -97,11 +97,11 @@ func TestHeaderTraversalMismatchedProviderStateError(t *testing.T) {
 	client := new(MockEthClient)
 
 	// start from genesis
-	headerTraversal := NewHeaderTraversal(client, nil)
+	headerTraversal := NewHeaderTraversal(client, nil, bigZero)
 
 	// blocks [0..4]
 	headers := makeHeaders(5, nil)
-	client.On("FinalizedBlockHeight").Return(big.NewInt(4), nil).Times(1) // Times so that we can override next
+	client.On("BlockHeaderByNumber", (*big.Int)(nil)).Return(&headers[4], nil).Times(1) // Times so that we can override next
 	client.On("BlockHeadersByRange", mock.MatchedBy(BigIntMatcher(0)), mock.MatchedBy(BigIntMatcher(4))).Return(headers, nil)
 	headers, err := headerTraversal.NextFinalizedHeaders(5)
 	require.NoError(t, err)
@@ -109,7 +109,7 @@ func TestHeaderTraversalMismatchedProviderStateError(t *testing.T) {
 
 	// blocks [5..9]. Next batch is not chained correctly (starts again from genesis)
 	headers = makeHeaders(5, nil)
-	client.On("FinalizedBlockHeight").Return(big.NewInt(9), nil)
+	client.On("BlockHeaderByNumber", (*big.Int)(nil)).Return(&types.Header{Number: big.NewInt(9)}, nil)
 	client.On("BlockHeadersByRange", mock.MatchedBy(BigIntMatcher(5)), mock.MatchedBy(BigIntMatcher(9))).Return(headers, nil)
 	headers, err = headerTraversal.NextFinalizedHeaders(5)
 	require.Nil(t, headers)

indexer/node/mocks.go

@@ -20,9 +20,9 @@ func (m *MockEthClient) BlockHeaderByNumber(number *big.Int) (*types.Header, err
 	return args.Get(0).(*types.Header), args.Error(1)
 }
 
-func (m *MockEthClient) FinalizedBlockHeight() (*big.Int, error) {
-	args := m.Called()
-	return args.Get(0).(*big.Int), args.Error(1)
+func (m *MockEthClient) BlockHeaderByHash(hash common.Hash) (*types.Header, error) {
+	args := m.Called(hash)
+	return args.Get(0).(*types.Header), args.Error(1)
 }
 
 func (m *MockEthClient) BlockHeadersByRange(from, to *big.Int) ([]types.Header, error) {
@@ -30,11 +30,6 @@ func (m *MockEthClient) BlockHeadersByRange(from, to *big.Int) ([]types.Header,
 	return args.Get(0).([]types.Header), args.Error(1)
 }
 
-func (m *MockEthClient) BlockHeaderByHash(hash common.Hash) (*types.Header, error) {
-	args := m.Called(hash)
-	return args.Get(0).(*types.Header), args.Error(1)
-}
-
 func (m *MockEthClient) StorageHash(address common.Address, blockNumber *big.Int) (common.Hash, error) {
 	args := m.Called(address, blockNumber)
 	return args.Get(0).(common.Hash), args.Error(1)