op-service: Refactor EthClient with ReceiptsProvider abstraction

op-service/sources/eth_client.go

@@ -96,21 +96,14 @@ func (c *EthClientConfig) Check() error {
 type EthClient struct {
 	client client.RPC
 
-	maxBatchSize int
+	recProvider *CachingReceiptsProvider
 
 	trustRPC bool
 
 	mustBePostMerge bool
 
-	provKind RPCProviderKind
-
 	log log.Logger
 
-	// cache receipts in bundles per block hash
-	// We cache the receipts fetching job to not lose progress when we have to retry the `Fetch` call
-	// common.Hash -> *receiptsFetchingJob
-	receiptsCache *caching.LRUCache[common.Hash, *receiptsFetchingJob]
-
 	// cache transactions in bundles per block hash
 	// common.Hash -> types.Transactions
 	transactionsCache *caching.LRUCache[common.Hash, types.Transactions]
@@ -122,46 +115,6 @@ type EthClient struct {
 	// cache payloads by hash
 	// common.Hash -> *eth.ExecutionPayload
 	payloadsCache *caching.LRUCache[common.Hash, *eth.ExecutionPayload]
-
-	// availableReceiptMethods tracks which receipt methods can be used for fetching receipts
-	// This may be modified concurrently, but we don't lock since it's a single
-	// uint64 that's not critical (fine to miss or mix up a modification)
-	availableReceiptMethods ReceiptsFetchingMethod
-
-	// lastMethodsReset tracks when availableReceiptMethods was last reset.
-	// When receipt-fetching fails it falls back to available methods,
-	// but periodically it will try to reset to the preferred optimal methods.
-	lastMethodsReset time.Time
-
-	// methodResetDuration defines how long we take till we reset lastMethodsReset
-	methodResetDuration time.Duration
-
-	// [OPTIONAL] The reth DB path to fetch receipts from
-	rethDbPath string
-}
-
-func (s *EthClient) PickReceiptsMethod(txCount uint64) ReceiptsFetchingMethod {
-	if now := time.Now(); now.Sub(s.lastMethodsReset) > s.methodResetDuration {
-		m := AvailableReceiptsFetchingMethods(s.provKind)
-		if s.availableReceiptMethods != m {
-			s.log.Warn("resetting back RPC preferences, please review RPC provider kind setting", "kind", s.provKind.String())
-		}
-		s.availableReceiptMethods = m
-		s.lastMethodsReset = now
-	}
-	return PickBestReceiptsFetchingMethod(s.provKind, s.availableReceiptMethods, txCount)
-}
-
-func (s *EthClient) OnReceiptsMethodErr(m ReceiptsFetchingMethod, err error) {
-	if unusableMethod(err) {
-		// clear the bit of the method that errored
-		s.availableReceiptMethods &^= m
-		s.log.Warn("failed to use selected RPC method for receipt fetching, temporarily falling back to alternatives",
-			"provider_kind", s.provKind, "failed_method", m, "fallback", s.availableReceiptMethods, "err", err)
-	} else {
-		s.log.Debug("failed to use selected RPC method for receipt fetching, but method does appear to be available, so we continue to use it",
-			"provider_kind", s.provKind, "failed_method", m, "fallback", s.availableReceiptMethods&^m, "err", err)
-	}
 }
 
 // NewEthClient returns an [EthClient], wrapping an RPC with bindings to fetch ethereum data with added error logging,
@@ -170,22 +123,32 @@ func NewEthClient(client client.RPC, log log.Logger, metrics caching.Metrics, co
 	if err := config.Check(); err != nil {
 		return nil, fmt.Errorf("bad config, cannot create L1 source: %w", err)
 	}
+	recCfg := RPCReceiptsConfig{
+		MaxBatchSize:        config.MaxRequestsPerBatch,
+		ProviderKind:        config.RPCProviderKind,
+		MethodResetDuration: config.MethodResetDuration,
+	}
+
 	client = LimitRPC(client, config.MaxConcurrentRequests)
+	recProvider := NewCachingRPCReceiptsProvider(client, log, recCfg, metrics, config.ReceiptsCacheSize)
+	// TODO: Need a way to create an EthClient with a RethDBReceiptsFetcher. Multiple options
+	//   1. If config contains rethdb path, favor creating it here instead of the RPCReceipsFetcher
+	//   2. Create new contstructor for EthClient
+	//      a. either specifically for RethDB or
+	//	    b. one where a receipts provider can be injected into
+	//     Option 2. has the benefit that this could all be completely hidden by
+	//     the rethdb build flag. Although some more work would need to be done to achieve this
+	//     higher up the call stack.
+	//  We might also want to move the receipts fetcher config out of the EthClientConfig.
 	return &EthClient{
-		client:                  client,
-		maxBatchSize:            config.MaxRequestsPerBatch,
-		trustRPC:                config.TrustRPC,
-		mustBePostMerge:         config.MustBePostMerge,
-		provKind:                config.RPCProviderKind,
-		log:                     log,
-		receiptsCache:           caching.NewLRUCache[common.Hash, *receiptsFetchingJob](metrics, "receipts", config.ReceiptsCacheSize),
-		transactionsCache:       caching.NewLRUCache[common.Hash, types.Transactions](metrics, "txs", config.TransactionsCacheSize),
-		headersCache:            caching.NewLRUCache[common.Hash, eth.BlockInfo](metrics, "headers", config.HeadersCacheSize),
-		payloadsCache:           caching.NewLRUCache[common.Hash, *eth.ExecutionPayload](metrics, "payloads", config.PayloadsCacheSize),
-		availableReceiptMethods: AvailableReceiptsFetchingMethods(config.RPCProviderKind),
-		lastMethodsReset:        time.Now(),
-		methodResetDuration:     config.MethodResetDuration,
-		rethDbPath:              config.RethDBPath,
+		client:            client,
+		recProvider:       recProvider,
+		trustRPC:          config.TrustRPC,
+		mustBePostMerge:   config.MustBePostMerge,
+		log:               log,
+		transactionsCache: caching.NewLRUCache[common.Hash, types.Transactions](metrics, "txs", config.TransactionsCacheSize),
+		headersCache:      caching.NewLRUCache[common.Hash, eth.BlockInfo](metrics, "headers", config.HeadersCacheSize),
+		payloadsCache:     caching.NewLRUCache[common.Hash, *eth.ExecutionPayload](metrics, "payloads", config.PayloadsCacheSize),
 	}, nil
 }
 
@@ -352,26 +315,31 @@ func (s *EthClient) PayloadByLabel(ctx context.Context, label eth.BlockLabel) (*
 // It verifies the receipt hash in the block header against the receipt hash of the fetched receipts
 // to ensure that the execution engine did not fail to return any receipts.
 func (s *EthClient) FetchReceipts(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Receipts, error) {
+	receipts, ok := s.recProvider.CachedReceipts(blockHash)
+	if ok {
+		// should always hit the cache
+		info, err := s.InfoByHash(ctx, blockHash)
+		if err != nil {
+			return nil, nil, fmt.Errorf("querying header: %w", err)
+		}
+		return info, receipts, nil
+	}
+
 	info, txs, err := s.InfoAndTxsByHash(ctx, blockHash)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, fmt.Errorf("querying block: %w", err)
 	}
-	// Try to reuse the receipts fetcher because is caches the results of intermediate calls. This means
-	// that if just one of many calls fail, we only retry the failed call rather than all of the calls.
-	// The underlying fetcher uses the receipts hash to verify receipt integrity.
-	var job *receiptsFetchingJob
-	if v, ok := s.receiptsCache.Get(blockHash); ok {
-		job = v
-	} else {
-		txHashes := eth.TransactionsToHashes(txs)
-		job = NewReceiptsFetchingJob(s, s.client, s.maxBatchSize, eth.ToBlockID(info), info.ReceiptHash(), txHashes, s.rethDbPath)
-		s.receiptsCache.Add(blockHash, job)
-	}
-	receipts, err := job.Fetch(ctx)
+
+	txHashes, block := eth.TransactionsToHashes(txs), eth.ToBlockID(info)
+	receipts, err = s.recProvider.FetchReceipts(ctx, block, txHashes)
 	if err != nil {
 		return nil, nil, err
 	}
 
+	if err := validateReceipts(block, info.ReceiptHash(), txHashes, receipts); err != nil {
+		return info, nil, fmt.Errorf("invalid receipts: %w", err)
+	}
+
 	return info, receipts, nil
 }