feature(op-node): pre-fetch receipts concurrently (ethereum-optimism#100

) * feature(op-node): concurrent pre-fetch receipts * use background ctx in GoOrUpdatePreFetchReceipts * change MaxConcurrentRequests from 10 to 20 --------- Co-authored-by: Welkin <[email protected]>
node-real · Dec 21, 2023 · 7947c25 · 7947c25
1 parent e62988a
commit 7947c25
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 7 deletions.
diff --git a/op-node/rollup/derive/engine_queue.go b/op-node/rollup/derive/engine_queue.go
@@ -787,7 +787,7 @@ func (eq *EngineQueue) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.System
 	if err != nil {
 		return NewTemporaryError(fmt.Errorf("failed to fetch L1 config of L2 block %s: %w", pipelineL2.ID(), err))
 	}
-	err2 := eq.l1Fetcher.GoOrUpdatePreFetchReceipts(ctx, pipelineOrigin.Number)
+	err2 := eq.l1Fetcher.GoOrUpdatePreFetchReceipts(context.Background(), pipelineOrigin.Number)
 	if err2 != nil {
 		return NewTemporaryError(fmt.Errorf("failed to run pre fetch L1 receipts for L1 start block %s: %w", pipelineOrigin.ID(), err2))
 	}

diff --git a/op-node/sources/l1_client.go b/op-node/sources/l1_client.go
@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
+	"golang.org/x/time/rate"
 
 	"github.com/ethereum-optimism/optimism/op-node/client"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
@@ -38,7 +39,7 @@ func L1ClientDefaultConfig(config *rollup.Config, trustRPC bool, kind RPCProvide
 			HeadersCacheSize:      span,
 			PayloadsCacheSize:     span,
 			MaxRequestsPerBatch:   20, // TODO: tune batch param
-			MaxConcurrentRequests: 10,
+			MaxConcurrentRequests: 20,
 			TrustRPC:              trustRPC,
 			MustBePostMerge:       false,
 			RPCProviderKind:       kind,
@@ -61,6 +62,8 @@ type L1Client struct {
 
 	//ensure pre-fetch receipts only once
 	preFetchReceiptsOnce sync.Once
+	//control the number of concurrent pre-fetch receipt requests.
+	preFetchReceiptsRateLimiter *rate.Limiter
 	//start block for pre-fetch receipts
 	preFetchReceiptsStartBlockChan chan uint64
 	//done chan
@@ -79,6 +82,7 @@ func NewL1Client(client client.RPC, log log.Logger, metrics caching.Metrics, con
 		l1BlockRefsCache:               caching.NewLRUCache(metrics, "blockrefs", config.L1BlockRefsCacheSize),
 		preFetchReceiptsOnce:           sync.Once{},
 		preFetchReceiptsStartBlockChan: make(chan uint64, 1),
+		preFetchReceiptsRateLimiter:    rate.NewLimiter(rate.Limit(config.MaxConcurrentRequests/2), config.MaxConcurrentRequests/2),
 		done:                           make(chan struct{}),
 	}, nil
 }
@@ -151,13 +155,20 @@ func (s *L1Client) GoOrUpdatePreFetchReceipts(ctx context.Context, l1Start uint6
 						time.Sleep(3 * time.Second)
 						continue
 					}
-					_, _, err = s.FetchReceipts(ctx, blockInfo.Hash)
-					if err != nil {
-						s.log.Warn("failed to pre-fetch receipts", "err", err)
-						time.Sleep(200 * time.Millisecond)
+					waitErr := s.preFetchReceiptsRateLimiter.Wait(ctx)
+					if waitErr != nil {
+						s.log.Warn("failed to wait pre-fetch receipts rateLimiter", "err", waitErr)
 						continue
 					}
-					s.log.Debug("pre-fetching receipts", "block", currentL1Block)
+
+					go func(ctx context.Context, blockInfo eth.L1BlockRef) {
+						_, _, err = s.FetchReceipts(ctx, blockInfo.Hash)
+						if err != nil {
+							s.log.Warn("failed to pre-fetch receipts", "err", err)
+							return
+						}
+						s.log.Debug("pre-fetching receipts", "block", currentL1Block)
+					}(ctx, blockInfo)
 					currentL1Block = currentL1Block + 1
 				}
 			}