Merge pull request #5505 from ethereum-optimism/aj/fpp-retry

op-program: Add retrying to L1 and L2 fetchers

op-program/host/host.go

@@ -79,7 +79,7 @@ func FaultProofProgram(logger log.Logger, cfg *config.Config) error {
 		l2DebugCl := &L2Source{L2Client: l2Cl, DebugClient: sources.NewDebugClient(l2RPC.CallContext)}
 
 		logger.Info("Setting up pre-fetcher")
-		prefetch := prefetcher.NewPrefetcher(l1Cl, l2DebugCl, kv)
+		prefetch := prefetcher.NewPrefetcher(logger, l1Cl, l2DebugCl, kv)
 		preimageOracle = asOracleFn(func(key common.Hash) ([]byte, error) {
 			return prefetch.GetPreimage(ctx, key)
 		})

op-program/host/prefetcher/prefetcher.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-program/client/l1"
@@ -38,10 +39,10 @@ type Prefetcher struct {
 	kvStore   kvstore.KV
 }
 
-func NewPrefetcher(l1Fetcher L1Source, l2Fetcher L2Source, kvStore kvstore.KV) *Prefetcher {
+func NewPrefetcher(logger log.Logger, l1Fetcher L1Source, l2Fetcher L2Source, kvStore kvstore.KV) *Prefetcher {
 	return &Prefetcher{
-		l1Fetcher: l1Fetcher,
-		l2Fetcher: l2Fetcher,
+		l1Fetcher: NewRetryingL1Source(logger, l1Fetcher),
+		l2Fetcher: NewRetryingL2Source(logger, l2Fetcher),
 		kvStore:   kvStore,
 	}
 }

op-program/host/prefetcher/prefetcher_test.go

@@ -5,9 +5,11 @@ import (
 	"math/rand"
 	"testing"
 
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/stretchr/testify/require"
 
@@ -263,6 +265,7 @@ type l2Client struct {
 }
 
 func createPrefetcher(t *testing.T) (*Prefetcher, *testutils.MockL1Source, *l2Client, kvstore.KV) {
+	logger := testlog.Logger(t, log.LvlDebug)
 	kv := kvstore.NewMemKV()
 
 	l1Source := new(testutils.MockL1Source)
@@ -271,7 +274,7 @@ func createPrefetcher(t *testing.T) (*Prefetcher, *testutils.MockL1Source, *l2Cl
 		MockDebugClient: new(testutils.MockDebugClient),
 	}
 
-	prefetcher := NewPrefetcher(l1Source, l2Source, kv)
+	prefetcher := NewPrefetcher(logger, l1Source, l2Source, kv)
 	return prefetcher, l1Source, l2Source, kv
 }
 

op-program/host/prefetcher/retry.go

+package prefetcher
+
+import (
+	"context"
+	"math"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const maxAttempts = math.MaxInt // Succeed or die trying
+
+type RetryingL1Source struct {
+	logger   log.Logger
+	source   L1Source
+	strategy backoff.Strategy
+}
+
+func NewRetryingL1Source(logger log.Logger, source L1Source) *RetryingL1Source {
+	return &RetryingL1Source{
+		logger:   logger,
+		source:   source,
+		strategy: backoff.Exponential(),
+	}
+}
+
+func (s *RetryingL1Source) InfoByHash(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, error) {
+	var info eth.BlockInfo
+	err := backoff.DoCtx(ctx, maxAttempts, s.strategy, func() error {
+		res, err := s.source.InfoByHash(ctx, blockHash)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve info", "hash", blockHash, "err", err)
+			return err
+		}
+		info = res
+		return nil
+	})
+	return info, err
+}
+
+func (s *RetryingL1Source) InfoAndTxsByHash(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Transactions, error) {
+	var info eth.BlockInfo
+	var txs types.Transactions
+	err := backoff.DoCtx(ctx, maxAttempts, s.strategy, func() error {
+		i, t, err := s.source.InfoAndTxsByHash(ctx, blockHash)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve info and txs", "hash", blockHash, "err", err)
+			return err
+		}
+		info = i
+		txs = t
+		return nil
+	})
+	return info, txs, err
+}
+
+func (s *RetryingL1Source) FetchReceipts(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Receipts, error) {
+	var info eth.BlockInfo
+	var rcpts types.Receipts
+	err := backoff.DoCtx(ctx, maxAttempts, s.strategy, func() error {
+		i, r, err := s.source.FetchReceipts(ctx, blockHash)
+		if err != nil {
+			s.logger.Warn("Failed to fetch receipts", "hash", blockHash, "err", err)
+			return err
+		}
+		info = i
+		rcpts = r
+		return nil
+	})
+	return info, rcpts, err
+}
+
+var _ L1Source = (*RetryingL1Source)(nil)
+
+type RetryingL2Source struct {
+	logger   log.Logger
+	source   L2Source
+	strategy backoff.Strategy
+}
+
+func (s *RetryingL2Source) InfoAndTxsByHash(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Transactions, error) {
+	var info eth.BlockInfo
+	var txs types.Transactions
+	err := backoff.DoCtx(ctx, maxAttempts, s.strategy, func() error {
+		i, t, err := s.source.InfoAndTxsByHash(ctx, blockHash)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve info and txs", "hash", blockHash, "err", err)
+			return err
+		}
+		info = i
+		txs = t
+		return nil
+	})
+	return info, txs, err
+}
+
+func (s *RetryingL2Source) NodeByHash(ctx context.Context, hash common.Hash) ([]byte, error) {
+	var node []byte
+	err := backoff.DoCtx(ctx, maxAttempts, s.strategy, func() error {
+		n, err := s.source.NodeByHash(ctx, hash)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve node", "hash", hash, "err", err)
+			return err
+		}
+		node = n
+		return nil
+	})
+	return node, err
+}
+
+func (s *RetryingL2Source) CodeByHash(ctx context.Context, hash common.Hash) ([]byte, error) {
+	var code []byte
+	err := backoff.DoCtx(ctx, maxAttempts, s.strategy, func() error {
+		c, err := s.source.CodeByHash(ctx, hash)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve code", "hash", hash, "err", err)
+			return err
+		}
+		code = c
+		return nil
+	})
+	return code, err
+}
+
+func NewRetryingL2Source(logger log.Logger, source L2Source) *RetryingL2Source {
+	return &RetryingL2Source{
+		logger:   logger,
+		source:   source,
+		strategy: backoff.Exponential(),
+	}
+}
+
+var _ L2Source = (*RetryingL2Source)(nil)

op-program/host/prefetcher/retry_test.go

+package prefetcher
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum-optimism/optimism/op-node/testutils"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRetryingL1Source(t *testing.T) {
+	ctx := context.Background()
+	hash := common.Hash{0xab}
+	info := &testutils.MockBlockInfo{InfoHash: hash}
+	// The mock really doesn't like returning nil for a eth.BlockInfo so return a value we expect to be ignored instead
+	wrongInfo := &testutils.MockBlockInfo{InfoHash: common.Hash{0x99}}
+	txs := types.Transactions{
+		&types.Transaction{},
+	}
+	rcpts := types.Receipts{
+		&types.Receipt{},
+	}
+
+	t.Run("InfoByHash Success", func(t *testing.T) {
+		source, mock := createL1Source(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectInfoByHash(hash, info, nil)
+
+		result, err := source.InfoByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, result)
+	})
+
+	t.Run("InfoByHash Error", func(t *testing.T) {
+		source, mock := createL1Source(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectInfoByHash(hash, wrongInfo, expectedErr)
+		mock.ExpectInfoByHash(hash, info, nil)
+
+		result, err := source.InfoByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, result)
+	})
+
+	t.Run("InfoAndTxsByHash Success", func(t *testing.T) {
+		source, mock := createL1Source(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectInfoAndTxsByHash(hash, info, txs, nil)
+
+		actualInfo, actualTxs, err := source.InfoAndTxsByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, actualInfo)
+		require.Equal(t, txs, actualTxs)
+	})
+
+	t.Run("InfoAndTxsByHash Error", func(t *testing.T) {
+		source, mock := createL1Source(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectInfoAndTxsByHash(hash, wrongInfo, nil, expectedErr)
+		mock.ExpectInfoAndTxsByHash(hash, info, txs, nil)
+
+		actualInfo, actualTxs, err := source.InfoAndTxsByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, actualInfo)
+		require.Equal(t, txs, actualTxs)
+	})
+
+	t.Run("FetchReceipts Success", func(t *testing.T) {
+		source, mock := createL1Source(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectFetchReceipts(hash, info, rcpts, nil)
+
+		actualInfo, actualRcpts, err := source.FetchReceipts(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, actualInfo)
+		require.Equal(t, rcpts, actualRcpts)
+	})
+
+	t.Run("FetchReceipts Error", func(t *testing.T) {
+		source, mock := createL1Source(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectFetchReceipts(hash, wrongInfo, nil, expectedErr)
+		mock.ExpectFetchReceipts(hash, info, rcpts, nil)
+
+		actualInfo, actualRcpts, err := source.FetchReceipts(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, actualInfo)
+		require.Equal(t, rcpts, actualRcpts)
+	})
+}
+
+func createL1Source(t *testing.T) (*RetryingL1Source, *testutils.MockL1Source) {
+	logger := testlog.Logger(t, log.LvlDebug)
+	mock := &testutils.MockL1Source{}
+	source := NewRetryingL1Source(logger, mock)
+	// Avoid sleeping in tests by using a fixed backoff strategy with no delay
+	source.strategy = backoff.Fixed(0)
+	return source, mock
+}
+
+func TestRetryingL2Source(t *testing.T) {
+	ctx := context.Background()
+	hash := common.Hash{0xab}
+	info := &testutils.MockBlockInfo{InfoHash: hash}
+	// The mock really doesn't like returning nil for a eth.BlockInfo so return a value we expect to be ignored instead
+	wrongInfo := &testutils.MockBlockInfo{InfoHash: common.Hash{0x99}}
+	txs := types.Transactions{
+		&types.Transaction{},
+	}
+	data := []byte{1, 2, 3, 4, 5}
+
+	t.Run("InfoAndTxsByHash Success", func(t *testing.T) {
+		source, mock := createL2Source(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectInfoAndTxsByHash(hash, info, txs, nil)
+
+		actualInfo, actualTxs, err := source.InfoAndTxsByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, actualInfo)
+		require.Equal(t, txs, actualTxs)
+	})
+
+	t.Run("InfoAndTxsByHash Error", func(t *testing.T) {
+		source, mock := createL2Source(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectInfoAndTxsByHash(hash, wrongInfo, nil, expectedErr)
+		mock.ExpectInfoAndTxsByHash(hash, info, txs, nil)
+
+		actualInfo, actualTxs, err := source.InfoAndTxsByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, info, actualInfo)
+		require.Equal(t, txs, actualTxs)
+	})
+
+	t.Run("NodeByHash Success", func(t *testing.T) {
+		source, mock := createL2Source(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectNodeByHash(hash, data, nil)
+
+		actual, err := source.NodeByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, data, actual)
+	})
+
+	t.Run("NodeByHash Error", func(t *testing.T) {
+		source, mock := createL2Source(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectNodeByHash(hash, nil, expectedErr)
+		mock.ExpectNodeByHash(hash, data, nil)
+
+		actual, err := source.NodeByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, data, actual)
+	})
+
+	t.Run("CodeByHash Success", func(t *testing.T) {
+		source, mock := createL2Source(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectCodeByHash(hash, data, nil)
+
+		actual, err := source.CodeByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, data, actual)
+	})
+
+	t.Run("CodeByHash Error", func(t *testing.T) {
+		source, mock := createL2Source(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectCodeByHash(hash, nil, expectedErr)
+		mock.ExpectCodeByHash(hash, data, nil)
+
+		actual, err := source.CodeByHash(ctx, hash)
+		require.NoError(t, err)
+		require.Equal(t, data, actual)
+	})
+}
+
+func createL2Source(t *testing.T) (*RetryingL2Source, *MockL2Source) {
+	logger := testlog.Logger(t, log.LvlDebug)
+	mock := &MockL2Source{}
+	source := NewRetryingL2Source(logger, mock)
+	// Avoid sleeping in tests by using a fixed backoff strategy with no delay
+	source.strategy = backoff.Fixed(0)
+	return source, mock
+}
+
+type MockL2Source struct {
+	mock.Mock
+}
+
+func (m *MockL2Source) InfoAndTxsByHash(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Transactions, error) {
+	out := m.Mock.MethodCalled("InfoAndTxsByHash", blockHash)
+	return out[0].(eth.BlockInfo), out[1].(types.Transactions), *out[2].(*error)
+}
+
+func (m *MockL2Source) NodeByHash(ctx context.Context, hash common.Hash) ([]byte, error) {
+	out := m.Mock.MethodCalled("NodeByHash", hash)
+	return out[0].([]byte), *out[1].(*error)
+}
+
+func (m *MockL2Source) CodeByHash(ctx context.Context, hash common.Hash) ([]byte, error) {
+	out := m.Mock.MethodCalled("CodeByHash", hash)
+	return out[0].([]byte), *out[1].(*error)
+}
+
+func (m *MockL2Source) ExpectInfoAndTxsByHash(blockHash common.Hash, info eth.BlockInfo, txs types.Transactions, err error) {
+	m.Mock.On("InfoAndTxsByHash", blockHash).Once().Return(info, txs, &err)
+}
+
+func (m *MockL2Source) ExpectNodeByHash(hash common.Hash, node []byte, err error) {
+	m.Mock.On("NodeByHash", hash).Once().Return(node, &err)
+}
+
+func (m *MockL2Source) ExpectCodeByHash(hash common.Hash, code []byte, err error) {
+	m.Mock.On("CodeByHash", hash).Once().Return(code, &err)
+}
+
+var _ L2Source = (*MockL2Source)(nil)