eth_client.go 14.5 KB
Newer Older
Andreas Bigger's avatar
Andreas Bigger committed
1 2 3 4 5 6 7 8 9
// Package sources exports a number of clients used to access ethereum chain data.
//
// There are a number of these exported clients used by the op-node:
// [L1Client] wraps an RPC client to retrieve L1 ethereum data.
// [L2Client] wraps an RPC client to retrieve L2 ethereum data.
// [RollupClient] wraps an RPC client to retrieve rollup data.
// [EngineClient] extends the [L2Client] providing engine API bindings.
//
// Internally, the listed clients wrap an [EthClient] which itself wraps a specified RPC client.
10
package sources
11 12 13 14

import (
	"context"
	"fmt"
15
	"math/big"
16

17
	"github.com/ethereum/go-ethereum"
18 19 20
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/common/hexutil"
	"github.com/ethereum/go-ethereum/core/types"
21
	"github.com/ethereum/go-ethereum/log"
22 23 24 25

	"github.com/ethereum-optimism/optimism/op-node/client"
	"github.com/ethereum-optimism/optimism/op-node/eth"
	"github.com/ethereum-optimism/optimism/op-node/sources/caching"
26 27
)

28
type EthClientConfig struct {
29
	// Maximum number of requests to make per batch
30 31 32 33 34 35 36 37 38 39 40 41 42
	MaxRequestsPerBatch int

	// limit concurrent requests, applies to the source as a whole
	MaxConcurrentRequests int

	// cache sizes

	// Number of blocks worth of receipts to cache
	ReceiptsCacheSize int
	// Number of blocks worth of transactions to cache
	TransactionsCacheSize int
	// Number of block headers to cache
	HeadersCacheSize int
43 44
	// Number of payloads to cache
	PayloadsCacheSize int
45 46 47 48 49 50

	// If the RPC is untrusted, then we should not use cached information from responses,
	// and instead verify against the block-hash.
	// Of real L1 blocks no deposits can be missed/faked, no batches can be missed/faked,
	// only the wrong L1 blocks can be retrieved.
	TrustRPC bool
51 52 53 54 55

	// If the RPC must ensure that the results fit the ExecutionPayload(Header) format.
	// If this is not checked, disabled header fields like the nonce or difficulty
	// may be used to get a different block-hash.
	MustBePostMerge bool
56 57 58

	// RPCProviderKind is a hint at what type of RPC provider we are dealing with
	RPCProviderKind RPCProviderKind
59 60
}

61
func (c *EthClientConfig) Check() error {
62 63 64 65 66 67 68 69 70
	if c.ReceiptsCacheSize < 0 {
		return fmt.Errorf("invalid receipts cache size: %d", c.ReceiptsCacheSize)
	}
	if c.TransactionsCacheSize < 0 {
		return fmt.Errorf("invalid transactions cache size: %d", c.TransactionsCacheSize)
	}
	if c.HeadersCacheSize < 0 {
		return fmt.Errorf("invalid headers cache size: %d", c.HeadersCacheSize)
	}
71 72 73
	if c.PayloadsCacheSize < 0 {
		return fmt.Errorf("invalid payloads cache size: %d", c.PayloadsCacheSize)
	}
74 75 76 77 78 79
	if c.MaxConcurrentRequests < 1 {
		return fmt.Errorf("expected at least 1 concurrent request, but max is %d", c.MaxConcurrentRequests)
	}
	if c.MaxRequestsPerBatch < 1 {
		return fmt.Errorf("expected at least 1 request per batch, but max is: %d", c.MaxRequestsPerBatch)
	}
80 81 82
	if !ValidRPCProviderKind(c.RPCProviderKind) {
		return fmt.Errorf("unknown rpc provider kind: %s", c.RPCProviderKind)
	}
83 84 85
	return nil
}

86 87
// EthClient retrieves ethereum data with optimized batch requests, cached results, and flag to not trust the RPC.
type EthClient struct {
88
	client client.RPC
89

90
	maxBatchSize int
91 92 93

	trustRPC bool

94 95
	mustBePostMerge bool

96 97
	provKind RPCProviderKind

98 99
	log log.Logger

100
	// cache receipts in bundles per block hash
101 102
	// We cache the receipts fetching job to not lose progress when we have to retry the `Fetch` call
	// common.Hash -> *receiptsFetchingJob
103
	receiptsCache *caching.LRUCache
104 105 106

	// cache transactions in bundles per block hash
	// common.Hash -> types.Transactions
107
	transactionsCache *caching.LRUCache
108 109 110

	// cache block headers of blocks by hash
	// common.Hash -> *HeaderInfo
111
	headersCache *caching.LRUCache
112 113 114 115

	// cache payloads by hash
	// common.Hash -> *eth.ExecutionPayload
	payloadsCache *caching.LRUCache
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136

	// 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
}

func (s *EthClient) PickReceiptsMethod(txCount uint64) ReceiptsFetchingMethod {
	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, 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)
	}
137 138
}

Andreas Bigger's avatar
Andreas Bigger committed
139 140
// NewEthClient returns an [EthClient], wrapping an RPC with bindings to fetch ethereum data with added error logging,
// metric tracking, and caching. The [EthClient] uses a [LimitRPC] wrapper to limit the number of concurrent RPC requests.
141
func NewEthClient(client client.RPC, log log.Logger, metrics caching.Metrics, config *EthClientConfig) (*EthClient, error) {
142 143 144 145
	if err := config.Check(); err != nil {
		return nil, fmt.Errorf("bad config, cannot create L1 source: %w", err)
	}
	client = LimitRPC(client, config.MaxConcurrentRequests)
146
	return &EthClient{
147 148 149 150 151 152 153 154 155 156 157
		client:                  client,
		maxBatchSize:            config.MaxRequestsPerBatch,
		trustRPC:                config.TrustRPC,
		mustBePostMerge:         config.MustBePostMerge,
		provKind:                config.RPCProviderKind,
		log:                     log,
		receiptsCache:           caching.NewLRUCache(metrics, "receipts", config.ReceiptsCacheSize),
		transactionsCache:       caching.NewLRUCache(metrics, "txs", config.TransactionsCacheSize),
		headersCache:            caching.NewLRUCache(metrics, "headers", config.HeadersCacheSize),
		payloadsCache:           caching.NewLRUCache(metrics, "payloads", config.PayloadsCacheSize),
		availableReceiptMethods: AvailableReceiptsFetchingMethods(config.RPCProviderKind),
158 159 160 161
	}, nil
}

// SubscribeNewHead subscribes to notifications about the current blockchain head on the given channel.
162
func (s *EthClient) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
163 164 165 166 167
	// Note that *types.Header does not cache the block hash unlike *HeaderInfo, it always recomputes.
	// Inefficient if used poorly, but no trust issue.
	return s.client.EthSubscribe(ctx, ch, "newHeads")
}

168
func (s *EthClient) headerCall(ctx context.Context, method string, id any) (*HeaderInfo, error) {
169 170 171 172 173 174 175 176
	var header *rpcHeader
	err := s.client.CallContext(ctx, &header, method, id, false) // headers are just blocks without txs
	if err != nil {
		return nil, err
	}
	if header == nil {
		return nil, ethereum.NotFound
	}
177
	info, err := header.Info(s.trustRPC, s.mustBePostMerge)
178 179 180
	if err != nil {
		return nil, err
	}
181
	s.headersCache.Add(info.Hash(), info)
182 183 184
	return info, nil
}

185
func (s *EthClient) blockCall(ctx context.Context, method string, id any) (*HeaderInfo, types.Transactions, error) {
186 187 188 189 190 191 192 193
	var block *rpcBlock
	err := s.client.CallContext(ctx, &block, method, id, true)
	if err != nil {
		return nil, nil, err
	}
	if block == nil {
		return nil, nil, ethereum.NotFound
	}
194
	info, txs, err := block.Info(s.trustRPC, s.mustBePostMerge)
195 196 197
	if err != nil {
		return nil, nil, err
	}
198 199
	s.headersCache.Add(info.Hash(), info)
	s.transactionsCache.Add(info.Hash(), txs)
200 201 202
	return info, txs, nil
}

203
func (s *EthClient) payloadCall(ctx context.Context, method string, id any) (*eth.ExecutionPayload, error) {
204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219
	var block *rpcBlock
	err := s.client.CallContext(ctx, &block, method, id, true)
	if err != nil {
		return nil, err
	}
	if block == nil {
		return nil, ethereum.NotFound
	}
	payload, err := block.ExecutionPayload(s.trustRPC)
	if err != nil {
		return nil, err
	}
	s.payloadsCache.Add(payload.BlockHash, payload)
	return payload, nil
}

220 221
// ChainID fetches the chain id of the internal RPC.
func (s *EthClient) ChainID(ctx context.Context) (*big.Int, error) {
Andreas Bigger's avatar
Andreas Bigger committed
222
	var id hexutil.Big
223 224 225 226
	err := s.client.CallContext(ctx, &id, "eth_chainId")
	if err != nil {
		return nil, err
	}
Andreas Bigger's avatar
Andreas Bigger committed
227
	return (*big.Int)(&id), nil
228 229
}

230
func (s *EthClient) InfoByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, error) {
231 232 233 234 235 236
	if header, ok := s.headersCache.Get(hash); ok {
		return header.(*HeaderInfo), nil
	}
	return s.headerCall(ctx, "eth_getBlockByHash", hash)
}

237
func (s *EthClient) InfoByNumber(ctx context.Context, number uint64) (eth.BlockInfo, error) {
238 239 240 241
	// can't hit the cache when querying by number due to reorgs.
	return s.headerCall(ctx, "eth_getBlockByNumber", hexutil.EncodeUint64(number))
}

242 243 244 245 246 247
func (s *EthClient) InfoByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, error) {
	// can't hit the cache when querying the head due to reorgs / changes.
	return s.headerCall(ctx, "eth_getBlockByNumber", string(label))
}

func (s *EthClient) InfoAndTxsByHash(ctx context.Context, hash common.Hash) (eth.BlockInfo, types.Transactions, error) {
248 249 250 251 252 253 254 255
	if header, ok := s.headersCache.Get(hash); ok {
		if txs, ok := s.transactionsCache.Get(hash); ok {
			return header.(*HeaderInfo), txs.(types.Transactions), nil
		}
	}
	return s.blockCall(ctx, "eth_getBlockByHash", hash)
}

256
func (s *EthClient) InfoAndTxsByNumber(ctx context.Context, number uint64) (eth.BlockInfo, types.Transactions, error) {
257 258 259 260
	// can't hit the cache when querying by number due to reorgs.
	return s.blockCall(ctx, "eth_getBlockByNumber", hexutil.EncodeUint64(number))
}

261
func (s *EthClient) InfoAndTxsByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, types.Transactions, error) {
262
	// can't hit the cache when querying the head due to reorgs / changes.
263
	return s.blockCall(ctx, "eth_getBlockByNumber", string(label))
264 265
}

266 267 268 269 270 271 272 273 274 275 276 277 278 279 280
func (s *EthClient) PayloadByHash(ctx context.Context, hash common.Hash) (*eth.ExecutionPayload, error) {
	if payload, ok := s.payloadsCache.Get(hash); ok {
		return payload.(*eth.ExecutionPayload), nil
	}
	return s.payloadCall(ctx, "eth_getBlockByHash", hash)
}

func (s *EthClient) PayloadByNumber(ctx context.Context, number uint64) (*eth.ExecutionPayload, error) {
	return s.payloadCall(ctx, "eth_getBlockByNumber", hexutil.EncodeUint64(number))
}

func (s *EthClient) PayloadByLabel(ctx context.Context, label eth.BlockLabel) (*eth.ExecutionPayload, error) {
	return s.payloadCall(ctx, "eth_getBlockByNumber", string(label))
}

281 282 283 284
// FetchReceipts returns a block info and all of the receipts associated with transactions in the block.
// 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) {
285
	info, txs, err := s.InfoAndTxsByHash(ctx, blockHash)
286
	if err != nil {
287
		return nil, nil, err
288
	}
289 290 291
	// 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.
292
	var job *receiptsFetchingJob
293
	if v, ok := s.receiptsCache.Get(blockHash); ok {
294
		job = v.(*receiptsFetchingJob)
295 296 297 298 299
	} else {
		txHashes := make([]common.Hash, len(txs))
		for i := 0; i < len(txs); i++ {
			txHashes[i] = txs[i].Hash()
		}
300 301
		job = NewReceiptsFetchingJob(s, s.client, s.maxBatchSize, eth.ToBlockID(info), info.ReceiptHash(), txHashes)
		s.receiptsCache.Add(blockHash, job)
302
	}
303
	receipts, err := job.Fetch(ctx)
304 305 306 307 308
	if err != nil {
		return nil, nil, err
	}

	return info, receipts, nil
309 310
}

311 312 313 314
// GetProof returns an account proof result, with any optional requested storage proofs.
// The retrieval does sanity-check that storage proofs for the expected keys are present in the response,
// but does not verify the result. Call accountResult.Verify(stateRoot) to verify the result.
func (s *EthClient) GetProof(ctx context.Context, address common.Address, storage []common.Hash, blockTag string) (*eth.AccountResult, error) {
315
	var getProofResponse *eth.AccountResult
316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360
	err := s.client.CallContext(ctx, &getProofResponse, "eth_getProof", address, storage, blockTag)
	if err != nil {
		return nil, err
	}
	if getProofResponse == nil {
		return nil, ethereum.NotFound
	}
	if len(getProofResponse.StorageProof) != len(storage) {
		return nil, fmt.Errorf("missing storage proof data, got %d proof entries but requested %d storage keys", len(getProofResponse.StorageProof), len(storage))
	}
	for i, key := range storage {
		if key != getProofResponse.StorageProof[i].Key {
			return nil, fmt.Errorf("unexpected storage proof key difference for entry %d: got %s but requested %s", i, getProofResponse.StorageProof[i].Key, key)
		}
	}
	return getProofResponse, nil
}

// GetStorageAt returns the storage value at the given address and storage slot, **without verifying the correctness of the result**.
// This should only ever be used as alternative to GetProof when the user opts in.
// E.g. Erigon L1 node users may have to use this, since Erigon does not support eth_getProof, see https://github.com/ledgerwatch/erigon/issues/1349
func (s *EthClient) GetStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockTag string) (common.Hash, error) {
	var out common.Hash
	err := s.client.CallContext(ctx, &out, "eth_getStorageAt", address, storageSlot, blockTag)
	return out, err
}

// ReadStorageAt is a convenience method to read a single storage value at the given slot in the given account.
// The storage slot value is verified against the state-root of the given block if we do not trust the RPC provider, or directly retrieved without proof if we do trust the RPC.
func (s *EthClient) ReadStorageAt(ctx context.Context, address common.Address, storageSlot common.Hash, blockHash common.Hash) (common.Hash, error) {
	if s.trustRPC {
		return s.GetStorageAt(ctx, address, storageSlot, blockHash.String())
	}
	block, err := s.InfoByHash(ctx, blockHash)
	if err != nil {
		return common.Hash{}, fmt.Errorf("failed to retrieve state root of block %s: %w", blockHash, err)
	}

	result, err := s.GetProof(ctx, address, []common.Hash{storageSlot}, blockHash.String())
	if err != nil {
		return common.Hash{}, fmt.Errorf("failed to fetch proof of storage slot %s at block %s: %w", storageSlot, blockHash, err)
	}

	if err := result.Verify(block.Root()); err != nil {
		return common.Hash{}, fmt.Errorf("failed to verify retrieved proof against state root: %w", err)
361
	}
Michael de Hoog's avatar
Michael de Hoog committed
362 363
	value := result.StorageProof[0].Value.ToInt()
	return common.BytesToHash(value.Bytes()), nil
364 365 366
}

func (s *EthClient) Close() {
367 368
	s.client.Close()
}