Merge pull request #8191 from ethereum-optimism/prefetching_ethclient

Add `PrefetchingEthClient`, which builds a cache of ethclient data for callers.

op-node/flags/flags.go

@@ -124,6 +124,18 @@ var (
 		EnvVars: prefixEnvVars("L1_HTTP_POLL_INTERVAL"),
 		Value:   time.Second * 12,
 	}
+	L1PrefetchingWindow = &cli.Uint64Flag{
+		Name:    "l1.prefetching-window",
+		Usage:   "Number of L1 blocks to prefetch in the background. Disabled if 0.",
+		EnvVars: prefixEnvVars("L1_PREFETCHING_WINDOW"),
+		Value:   0,
+	}
+	L1PrefetchingTimeout = &cli.DurationFlag{
+		Name:    "l1.prefetching-timeout",
+		Usage:   "Timeout for L1 prefetching. Disabled if 0.",
+		EnvVars: prefixEnvVars("L1_PREFETCHING_TIMEOUT"),
+		Value:   time.Second * 30,
+	}
 	L2EngineJWTSecret = &cli.StringFlag{
 		Name:        "l2.jwt-secret",
 		Usage:       "Path to JWT secret key. Keys are 32 bytes, hex encoded in a file. A new key will be generated if left empty.",
@@ -309,6 +321,8 @@ var optionalFlags = []cli.Flag{
 	L1RPCMaxBatchSize,
 	L1RPCMaxConcurrency,
 	L1HTTPPollInterval,
+	L1PrefetchingWindow,
+	L1PrefetchingTimeout,
 	L2EngineJWTSecret,
 	VerifierL1Confs,
 	SequencerEnabledFlag,

op-node/node/client.go

@@ -159,6 +159,14 @@ type L1EndpointConfig struct {
 	// It is recommended to use websockets or IPC for efficient following of the changing block.
 	// Setting this to 0 disables polling.
 	HttpPollInterval time.Duration
+
+	// PrefetchingWindow specifies the number of blocks to prefetch from the L1 RPC.
+	// Setting this to 0 disables prefetching.
+	PrefetchingWindow uint64
+
+	// PrefetchingTimeout specifies the timeout for prefetching from the L1 RPC.
+	// Setting this to 0 disables prefetching.
+	PrefetchingTimeout time.Duration
 }
 
 var _ L1EndpointSetup = (*L1EndpointConfig)(nil)
@@ -192,6 +200,8 @@ func (cfg *L1EndpointConfig) Setup(ctx context.Context, log log.Logger, rollupCf
 	rpcCfg := sources.L1ClientDefaultConfig(rollupCfg, cfg.L1TrustRPC, cfg.L1RPCKind)
 	rpcCfg.MaxRequestsPerBatch = cfg.BatchSize
 	rpcCfg.MaxConcurrentRequests = cfg.MaxConcurrency
+	rpcCfg.PrefetchingWindow = cfg.PrefetchingWindow
+	rpcCfg.PrefetchingTimeout = cfg.PrefetchingTimeout
 	return l1Node, rpcCfg, nil
 }
 

op-node/service.go

@@ -123,13 +123,15 @@ func NewConfig(ctx *cli.Context, log log.Logger) (*node.Config, error) {
 
 func NewL1EndpointConfig(ctx *cli.Context) *node.L1EndpointConfig {
 	return &node.L1EndpointConfig{
-		L1NodeAddr:       ctx.String(flags.L1NodeAddr.Name),
-		L1TrustRPC:       ctx.Bool(flags.L1TrustRPC.Name),
-		L1RPCKind:        sources.RPCProviderKind(strings.ToLower(ctx.String(flags.L1RPCProviderKind.Name))),
-		RateLimit:        ctx.Float64(flags.L1RPCRateLimit.Name),
-		BatchSize:        ctx.Int(flags.L1RPCMaxBatchSize.Name),
-		HttpPollInterval: ctx.Duration(flags.L1HTTPPollInterval.Name),
-		MaxConcurrency:   ctx.Int(flags.L1RPCMaxConcurrency.Name),
+		L1NodeAddr:         ctx.String(flags.L1NodeAddr.Name),
+		L1TrustRPC:         ctx.Bool(flags.L1TrustRPC.Name),
+		L1RPCKind:          sources.RPCProviderKind(strings.ToLower(ctx.String(flags.L1RPCProviderKind.Name))),
+		PrefetchingWindow:  ctx.Uint64(flags.L1PrefetchingWindow.Name),
+		PrefetchingTimeout: ctx.Duration(flags.L1PrefetchingTimeout.Name),
+		RateLimit:          ctx.Float64(flags.L1RPCRateLimit.Name),
+		BatchSize:          ctx.Int(flags.L1RPCMaxBatchSize.Name),
+		HttpPollInterval:   ctx.Duration(flags.L1HTTPPollInterval.Name),
+		MaxConcurrency:     ctx.Int(flags.L1RPCMaxConcurrency.Name),
 	}
 }
 

op-service/sources/l1_client.go

@@ -20,6 +20,8 @@ type L1ClientConfig struct {
 	EthClientConfig
 
 	L1BlockRefsCacheSize int
+	PrefetchingWindow    uint64
+	PrefetchingTimeout   time.Duration
 }
 
 func L1ClientDefaultConfig(config *rollup.Config, trustRPC bool, kind RPCProviderKind) *L1ClientConfig {
@@ -45,6 +47,8 @@ func L1ClientDefaultConfig(config *rollup.Config, trustRPC bool, kind RPCProvide
 		},
 		// Not bounded by span, to cover find-sync-start range fully for speedy recovery after errors.
 		L1BlockRefsCacheSize: fullSpan,
+		PrefetchingWindow:    0, // no prefetching by default
+		PrefetchingTimeout:   0, // no prefetching by default
 	}
 }
 
@@ -52,7 +56,7 @@ func L1ClientDefaultConfig(config *rollup.Config, trustRPC bool, kind RPCProvide
 // with optimized batch requests, cached results, and flag to not trust the RPC
 // (i.e. to verify all returned contents against corresponding block hashes).
 type L1Client struct {
-	*EthClient
+	EthClientInterface
 
 	// cache L1BlockRef by hash
 	// common.Hash -> eth.L1BlockRef
@@ -66,10 +70,23 @@ func NewL1Client(client client.RPC, log log.Logger, metrics caching.Metrics, con
 		return nil, err
 	}
 
+	var clientToUse EthClientInterface
+
+	if config.PrefetchingTimeout > 0 && config.PrefetchingWindow > 0 {
+		prefetchingEthClient, err := NewPrefetchingEthClient(ethClient, config.PrefetchingWindow, config.PrefetchingTimeout)
+		if err != nil {
+			return nil, err
+		}
+		clientToUse = prefetchingEthClient
+	} else {
+		clientToUse = ethClient
+	}
+
 	return &L1Client{
-		EthClient:        ethClient,
-		l1BlockRefsCache: caching.NewLRUCache[common.Hash, eth.L1BlockRef](metrics, "blockrefs", config.L1BlockRefsCacheSize),
+		EthClientInterface: clientToUse,
+		l1BlockRefsCache:   caching.NewLRUCache[common.Hash, eth.L1BlockRef](metrics, "blockrefs", config.L1BlockRefsCacheSize),
 	}, nil
+
 }
 
 // L1BlockRefByLabel returns the [eth.L1BlockRef] for the given block label.

op-service/sources/prefetching_eth_client.go

+package sources
+
+import (
+	"context"
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type EthClientInterface interface {
+	SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error)
+	ChainID(ctx context.Context) (*big.Int, error)
+	InfoByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, error)
+	InfoByNumber(ctx context.Context, number uint64) (eth.BlockInfo, error)
+	InfoByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, error)
+	InfoAndTxsByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, types.Transactions, error)
+	InfoAndTxsByNumber(ctx context.Context, number uint64) (eth.BlockInfo, types.Transactions, error)
+	InfoAndTxsByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, types.Transactions, error)
+	PayloadByHash(ctx context.Context, hash common.Hash) (*eth.ExecutionPayload, error)
+	PayloadByNumber(ctx context.Context, number uint64) (*eth.ExecutionPayload, error)
+	PayloadByLabel(ctx context.Context, label eth.BlockLabel) (*eth.ExecutionPayload, error)
+	FetchReceipts(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Receipts, error)
+	GetProof(ctx context.Context, address common.Address, storage []common.Hash, blockTag string) (*eth.AccountResult, error)
+	GetStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockTag string) (common.Hash, error)
+	ReadStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockHash common.Hash) (common.Hash, error)
+	Close()
+}
+
+type PrefetchingEthClient struct {
+	inner                 EthClientInterface
+	PrefetchingRange      uint64
+	PrefetchingTimeout    time.Duration
+	runningCtx            context.Context
+	runningCancel         context.CancelFunc
+	highestHeadRequesting uint64
+	highestHeadLock       sync.Mutex
+	wg                    *sync.WaitGroup // used for testing
+}
+
+// NewPrefetchingEthClient creates a new [PrefetchingEthClient] with the given underlying [EthClient]
+// and a prefetching range.
+func NewPrefetchingEthClient(inner EthClientInterface, prefetchingRange uint64, timeout time.Duration) (*PrefetchingEthClient, error) {
+	// Create a new context for the prefetching goroutines
+	runningCtx, runningCancel := context.WithCancel(context.Background())
+	return &PrefetchingEthClient{
+		inner:                 inner,
+		PrefetchingRange:      prefetchingRange,
+		PrefetchingTimeout:    timeout,
+		runningCtx:            runningCtx,
+		runningCancel:         runningCancel,
+		highestHeadRequesting: 0,
+	}, nil
+}
+
+func (p *PrefetchingEthClient) updateRequestingHead(start, end uint64) (newStart uint64, shouldFetch bool) {
+	// Acquire lock before reading/updating highestHeadRequesting
+	p.highestHeadLock.Lock()
+	defer p.highestHeadLock.Unlock()
+	if start <= p.highestHeadRequesting {
+		start = p.highestHeadRequesting + 1
+	}
+	if p.highestHeadRequesting < end {
+		p.highestHeadRequesting = end
+	}
+	return start, start <= end
+}
+
+func (p *PrefetchingEthClient) FetchWindow(start, end uint64) {
+	if p.wg != nil {
+		defer p.wg.Done()
+	}
+
+	start, shouldFetch := p.updateRequestingHead(start, end)
+	if !shouldFetch {
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(p.runningCtx, p.PrefetchingTimeout)
+	defer cancel()
+	for i := start; i <= end; i++ {
+		p.FetchBlockAndReceipts(ctx, i)
+	}
+}
+
+func (p *PrefetchingEthClient) FetchBlockAndReceipts(ctx context.Context, number uint64) {
+	blockInfo, _, err := p.inner.InfoAndTxsByNumber(ctx, number)
+	if err != nil {
+		return
+	}
+	_, _, _ = p.inner.FetchReceipts(ctx, blockInfo.Hash())
+}
+
+func (p *PrefetchingEthClient) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
+	return p.inner.SubscribeNewHead(ctx, ch)
+}
+
+func (p *PrefetchingEthClient) ChainID(ctx context.Context) (*big.Int, error) {
+	return p.inner.ChainID(ctx)
+}
+
+func (p *PrefetchingEthClient) InfoByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, error) {
+	// Fetch the block information for the given hash
+	blockInfo, err := p.inner.InfoByHash(ctx, hash)
+	if err != nil {
+		return blockInfo, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts starting from the block number of the fetched block
+	go p.FetchWindow(blockInfo.NumberU64()+1, blockInfo.NumberU64()+p.PrefetchingRange)
+
+	return blockInfo, nil
+}
+
+func (p *PrefetchingEthClient) InfoByNumber(ctx context.Context, number uint64) (eth.BlockInfo, error) {
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Trigger prefetching in the background
+	go p.FetchWindow(number+1, number+p.PrefetchingRange)
+
+	// Fetch the requested block
+	return p.inner.InfoByNumber(ctx, number)
+}
+
+func (p *PrefetchingEthClient) InfoByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, error) {
+	// Fetch the block information for the given label
+	blockInfo, err := p.inner.InfoByLabel(ctx, label)
+	if err != nil {
+		return blockInfo, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts starting from the block number of the fetched block
+	go p.FetchWindow(blockInfo.NumberU64()+1, blockInfo.NumberU64()+p.PrefetchingRange)
+
+	return blockInfo, nil
+}
+
+func (p *PrefetchingEthClient) InfoAndTxsByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, types.Transactions, error) {
+	// Fetch the block info and transactions for the requested hash
+	blockInfo, txs, err := p.inner.InfoAndTxsByHash(ctx, hash)
+	if err != nil {
+		return blockInfo, txs, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(blockInfo.NumberU64()+1, blockInfo.NumberU64()+p.PrefetchingRange)
+
+	return blockInfo, txs, nil
+}
+
+func (p *PrefetchingEthClient) InfoAndTxsByNumber(ctx context.Context, number uint64) (eth.BlockInfo, types.Transactions, error) {
+	// Fetch the block info and transactions for the requested number
+	blockInfo, txs, err := p.inner.InfoAndTxsByNumber(ctx, number)
+	if err != nil {
+		return blockInfo, txs, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(number+1, number+p.PrefetchingRange)
+
+	return blockInfo, txs, nil
+}
+
+func (p *PrefetchingEthClient) InfoAndTxsByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, types.Transactions, error) {
+	// Fetch the block info and transactions for the requested label
+	blockInfo, txs, err := p.inner.InfoAndTxsByLabel(ctx, label)
+	if err != nil {
+		return blockInfo, txs, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(blockInfo.NumberU64()+1, blockInfo.NumberU64()+p.PrefetchingRange)
+
+	return blockInfo, txs, nil
+}
+
+func (p *PrefetchingEthClient) PayloadByHash(ctx context.Context, hash common.Hash) (*eth.ExecutionPayload, error) {
+	// Fetch the payload for the requested hash
+	payload, err := p.inner.PayloadByHash(ctx, hash)
+	if err != nil {
+		return payload, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(uint64(payload.BlockNumber)+1, uint64(payload.BlockNumber)+p.PrefetchingRange)
+
+	return payload, nil
+}
+
+func (p *PrefetchingEthClient) PayloadByNumber(ctx context.Context, number uint64) (*eth.ExecutionPayload, error) {
+	// Fetch the payload for the requested number
+	payload, err := p.inner.PayloadByNumber(ctx, number)
+	if err != nil {
+		return payload, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(number+1, number+p.PrefetchingRange)
+
+	return payload, nil
+}
+
+func (p *PrefetchingEthClient) PayloadByLabel(ctx context.Context, label eth.BlockLabel) (*eth.ExecutionPayload, error) {
+	// Fetch the payload for the requested label
+	payload, err := p.inner.PayloadByLabel(ctx, label)
+	if err != nil {
+		return payload, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(uint64(payload.BlockNumber)+1, uint64(payload.BlockNumber)+p.PrefetchingRange)
+
+	return payload, nil
+}
+
+func (p *PrefetchingEthClient) FetchReceipts(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Receipts, error) {
+	// Fetch the block info and receipts for the requested hash
+	blockInfo, receipts, err := p.inner.FetchReceipts(ctx, blockHash)
+	if err != nil {
+		return blockInfo, receipts, err
+	}
+
+	if p.wg != nil {
+		p.wg.Add(1)
+	}
+	// Prefetch the next n blocks and their receipts
+	go p.FetchWindow(blockInfo.NumberU64(), blockInfo.NumberU64()+p.PrefetchingRange)
+
+	return blockInfo, receipts, nil
+}
+
+func (p *PrefetchingEthClient) GetProof(ctx context.Context, address common.Address, storage []common.Hash, blockTag string) (*eth.AccountResult, error) {
+	return p.inner.GetProof(ctx, address, storage, blockTag)
+}
+
+func (p *PrefetchingEthClient) GetStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockTag string) (common.Hash, error) {
+	return p.inner.GetStorageAt(ctx, address, storageSlot, blockTag)
+}
+
+func (p *PrefetchingEthClient) ReadStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockHash common.Hash) (common.Hash, error) {
+	return p.inner.ReadStorageAt(ctx, address, storageSlot, blockHash)
+}
+
+func (p *PrefetchingEthClient) Close() {
+	p.runningCancel()
+	p.inner.Close()
+}

op-service/sources/prefetching_eth_client_test.go

+package sources
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/testutils"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/stretchr/testify/require"
+)
+
+// TestPrefetchingEthClient runs all test cases for each prefetching range.
+func TestPrefetchingEthClient(t *testing.T) {
+	prefetchingRanges := []uint64{0, 1, 5}
+	for _, prefetchingRange := range prefetchingRanges {
+		testName := fmt.Sprintf("range-%d", prefetchingRange)
+		t.Run(testName, func(t *testing.T) {
+			ctx := context.Background()
+			mockEthClient := new(testutils.MockEthClient)
+			client, err := NewPrefetchingEthClient(mockEthClient, prefetchingRange, 30*time.Second)
+			require.NoError(t, err)
+			defer client.Close()
+			client.wg = new(sync.WaitGroup) // Initialize the WaitGroup for testing
+
+			// set up a random block to get from the client
+			randomness := rand.New(rand.NewSource(123))
+			block, _ := randomRpcBlockAndReceipts(randomness, 2)
+			rhdr := block.rpcHeader
+			expectedTxs := block.Transactions
+			expectedInfo, err := rhdr.Info(true, false)
+			require.NoError(t, err)
+			mockEthClient.ExpectInfoAndTxsByNumber(uint64(rhdr.Number), expectedInfo, expectedTxs, nil)
+
+			// also set up a window of random blocks and receipts to prefetch
+			windowEnd := (uint64(rhdr.Number) + client.PrefetchingRange)
+			for i := uint64(rhdr.Number) + 1; i <= windowEnd; i++ {
+				// set up different info per iteration
+				fillerBlock, fillerReceipts := randomRpcBlockAndReceipts(randomness, 2)
+				fillerBlock.rpcHeader.Number = hexutil.Uint64(i)
+				fillerInfo, err := fillerBlock.rpcHeader.Info(true, false)
+				require.NoError(t, err)
+				mockEthClient.ExpectInfoAndTxsByNumber(i, fillerInfo, fillerBlock.Transactions, nil)
+				mockEthClient.ExpectFetchReceipts(fillerBlock.Hash, fillerInfo, fillerReceipts, nil)
+			}
+
+			info, txs, err := client.InfoAndTxsByNumber(ctx, uint64(rhdr.Number))
+			require.NoError(t, err)
+			require.Equal(t, info, expectedInfo)
+			require.Equal(t, txs, types.Transactions(expectedTxs))
+			client.wg.Wait() // Wait for all goroutines to complete before asserting expectations
+			mockEthClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestUpdateRequestingHead_NormalRange(t *testing.T) {
+	client := &PrefetchingEthClient{
+		highestHeadRequesting: 10,
+		PrefetchingTimeout:    30 * time.Second,
+	}
+
+	start, end := uint64(11), uint64(15)
+	newStart, shouldFetch := client.updateRequestingHead(start, end)
+
+	require.Equal(t, newStart, start)
+	require.True(t, shouldFetch)
+	require.Equal(t, client.highestHeadRequesting, end)
+}
+
+func TestUpdateRequestingHead_OverlappingRange(t *testing.T) {
+	highestHeadBeforeUpdate := uint64(10)
+	client := &PrefetchingEthClient{
+		highestHeadRequesting: highestHeadBeforeUpdate,
+		PrefetchingTimeout:    30 * time.Second,
+	}
+
+	start, end := uint64(8), uint64(12)
+	newStart, shouldFetch := client.updateRequestingHead(start, end)
+
+	require.Equal(t, newStart, highestHeadBeforeUpdate+1)
+	require.True(t, shouldFetch)
+	require.Equal(t, client.highestHeadRequesting, end)
+}

op-service/testutils/mock_eth_client.go

@@ -2,9 +2,11 @@ package testutils
 
 import (
 	"context"
+	"math/big"
 
 	"github.com/stretchr/testify/mock"
 
+	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 
@@ -15,6 +17,15 @@ type MockEthClient struct {
 	mock.Mock
 }
 
+func (m *MockEthClient) ChainID(ctx context.Context) (*big.Int, error) {
+	out := m.Mock.MethodCalled("ChainID")
+	return out[0].(*big.Int), *out[1].(*error)
+}
+
+func (m *MockEthClient) ExpectChainID(chainID *big.Int, err error) {
+	m.Mock.On("ChainID").Once().Return(chainID, &err)
+}
+
 func (m *MockEthClient) InfoByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, error) {
 	out := m.Mock.MethodCalled("InfoByHash", hash)
 	return *out[0].(*eth.BlockInfo), *out[1].(*error)
@@ -128,3 +139,9 @@ func (m *MockEthClient) ReadStorageAt(ctx context.Context, address common.Addres
 func (m *MockEthClient) ExpectReadStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockHash common.Hash, result common.Hash, err error) {
 	m.Mock.On("ReadStorageAt", address, storageSlot, blockHash).Once().Return(result, &err)
 }
+
+func (m *MockEthClient) Close() {}
+
+func (m *MockEthClient) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
+	return nil, nil
+}