From 84f1e319091644af7bc4dffcc39d04099fce7507 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Thu, 13 Jun 2024 14:28:22 +0200 Subject: [PATCH] ethclient/lightclient: added comments and minor fixes --- ethclient/lightclient/blocks.go | 13 ++- ethclient/lightclient/canonical.go | 6 +- ethclient/lightclient/request_map.go | 112 ++++++++++++++++----- ethclient/lightclient/state.go | 52 +++++++++- ethclient/lightclient/transactions.go | 137 +++++++++++++++++++------- 5 files changed, 254 insertions(+), 66 deletions(-) diff --git a/ethclient/lightclient/blocks.go b/ethclient/lightclient/blocks.go index 800d337de9..8b2e0360ef 100644 --- a/ethclient/lightclient/blocks.go +++ b/ethclient/lightclient/blocks.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/log" ) +// blocksAndHeadersFields defines Client fields related to blocks and headers. type blocksAndHeadersFields struct { headerCache *lru.Cache[common.Hash, *types.Header] headerRequests *requestMap[common.Hash, *types.Header] @@ -38,6 +39,7 @@ type blocksAndHeadersFields struct { blockRequests *requestMap[common.Hash, *types.Block] } +// initBlocksAndHeaders initializes the structures related to blocks and headers. func (c *Client) initBlocksAndHeaders() { c.headerCache = lru.NewCache[common.Hash, *types.Header](1000) c.payloadHeaderCache = lru.NewCache[common.Hash, *btypes.ExecutionHeader](1000) @@ -46,6 +48,7 @@ func (c *Client) initBlocksAndHeaders() { c.blockRequests = newRequestMap[common.Hash, *types.Block](c.requestBlock) } +// closeBlocksAndHeaders shuts down the structures related to blocks and headers. func (c *Client) closeBlocksAndHeaders() { c.headerRequests.close() c.blockRequests.close() @@ -60,7 +63,7 @@ func (c *Client) getHeader(ctx context.Context, hash common.Hash) (*types.Header } if c.blockRequests.has(hash) && !c.headerRequests.has(hash) { req := c.blockRequests.request(hash) - block, err := req.getResult(ctx) + block, err := req.waitForResult(ctx) if err == nil { header := block.Header() c.headerCache.Add(hash, header) @@ -73,7 +76,7 @@ func (c *Client) getHeader(ctx context.Context, hash common.Hash) (*types.Header } } req := c.headerRequests.request(hash) - header, err := req.getResult(ctx) + header, err := req.waitForResult(ctx) if err == nil { c.headerCache.Add(hash, header) } @@ -95,7 +98,7 @@ func (c *Client) getBlock(ctx context.Context, hash common.Hash) (*types.Block, return block, nil } req := c.blockRequests.request(hash) - block, err := req.getResult(ctx) + block, err := req.waitForResult(ctx) if err == nil { header := block.Header() c.headerCache.Add(hash, header) @@ -107,6 +110,8 @@ func (c *Client) getBlock(ctx context.Context, hash common.Hash) (*types.Block, return block, err } +// requestHeader requests the header with the specified hash from the RPC client. +// Either the header with the given hash or an error is returned. func (c *Client) requestHeader(ctx context.Context, hash common.Hash) (*types.Header, error) { var header *types.Header log.Debug("Starting RPC request", "type", "eth_getBlockByHash", "hash", hash, "full", false) @@ -121,6 +126,8 @@ func (c *Client) requestHeader(ctx context.Context, hash common.Hash) (*types.He return header, err } +// requestBlock requests the block with the specified hash from the RPC client. +// Either the block with the given hash or an error is returned. func (c *Client) requestBlock(ctx context.Context, hash common.Hash) (*types.Block, error) { var ( raw json.RawMessage diff --git a/ethclient/lightclient/canonical.go b/ethclient/lightclient/canonical.go index a0d887d711..4c04f41272 100644 --- a/ethclient/lightclient/canonical.go +++ b/ethclient/lightclient/canonical.go @@ -33,6 +33,7 @@ import ( const recentCanonicalLength = 256 +// canonicalChainFields defines Client fields related to canonical chain tracking. type canonicalChainFields struct { chainLock sync.Mutex head, finality *btypes.ExecutionHeader @@ -43,6 +44,7 @@ type canonicalChainFields struct { requests *requestMap[uint64, common.Hash] // requested; neither recent nor cached finalized } +// initCanonicalChain initializes the structures related to canonical chain tracking. func (c *Client) initCanonicalChain() { c.finalized = lru.NewCache[uint64, common.Hash](10000) c.requests = newRequestMap[uint64, common.Hash](nil) @@ -51,6 +53,7 @@ func (c *Client) initCanonicalChain() { go c.tailFetcher() } +// closeCanonicalChain shuts down the structures related to canonical chain tracking. func (c *Client) closeCanonicalChain() { c.requests.close() close(c.closeCh) @@ -67,6 +70,7 @@ func (c *Client) setHead(head *btypes.ExecutionHeader) bool { if c.recent == nil || c.head == nil || c.head.BlockNumber()+1 != headNum || c.head.BlockHash() != head.ParentHash() { // initialize recent canonical hash map when first head is added or when // it is not a descendant of the previous head + c.clearStateCache() c.recent = make(map[uint64]common.Hash) if headNum > 0 { c.recent[headNum-1] = head.ParentHash() @@ -179,7 +183,7 @@ func (c *Client) getHash(ctx context.Context, number uint64) (common.Hash, error default: } defer req.release() - return req.getResult(ctx) + return req.waitForResult(ctx) } func (c *Client) getCachedHash(number uint64) (common.Hash, bool) { diff --git a/ethclient/lightclient/request_map.go b/ethclient/lightclient/request_map.go index 4682851f5d..c6070856a9 100644 --- a/ethclient/lightclient/request_map.go +++ b/ethclient/lightclient/request_map.go @@ -22,6 +22,11 @@ import ( "sync" ) +// requestMap maps pending requests onto a keyspace in order to avoid multiple +// processes requesting the same thing simultaneously. If a new item is requested +// by a process, it is added to the map and subsequest requesters of the same +// item will join the same request and receive the same result. An item stays in +// the map until it is explicitly removed or released by every requester. type requestMap[K comparable, V any] struct { lock sync.Mutex closed bool @@ -29,18 +34,25 @@ type requestMap[K comparable, V any] struct { requests map[K]*mappedRequest[K, V] } +// mappedRequest represents a request that multiple processes can join. type mappedRequest[K comparable, V any] struct { - lock sync.Mutex - rm *requestMap[K, V] - key K - refCount int - delivered, closed bool - deliveredCh chan struct{} - cancelFn func() // called when delivered || closed || refCount == 0 becomes true - result V - err error + lock sync.Mutex + rm *requestMap[K, V] // in the map if !closed && !removed && refCount != 0 + key K + refCount int + closed, delivered, removed bool + deliveredCh chan struct{} + cancelFn func() // called when closed || delivered || refCount == 0 becomes true + result V + err error } +// newRequestMap creates a new requestMap with the specified key and value types. +// If requestFn is specified then it is called by requestMap every time a new +// request is added to the map and the return value or error is automatically +// delivered. +// If requestFn == nil then the caller is expected to keep track of open requests +// and eventually deliver a result for them. func newRequestMap[K comparable, V any](requestFn func(context.Context, K) (V, error)) *requestMap[K, V] { return &requestMap[K, V]{ requestFn: requestFn, @@ -48,6 +60,11 @@ func newRequestMap[K comparable, V any](requestFn func(context.Context, K) (V, e } } +// close cancels all pending requests. Subsequent request attempts create dummy +// requests that are not sent or added to the map and instantly return an error +// result. This simplifies shutdown check on the caller side because there has +// to be an error check mechanism for request results anyway and there is no need +// to also check for errors at request creation. func (rm *requestMap[K, V]) close() { rm.lock.Lock() defer rm.lock.Unlock() @@ -55,19 +72,38 @@ func (rm *requestMap[K, V]) close() { if rm.closed { return } - for _, req := range rm.requests { - if !req.delivered && req.refCount != 0 { - req.cancelFn() - req.closed = true + for _, r := range rm.requests { + r.lock.Lock() + if !r.delivered && r.refCount != 0 { + r.cancelFn() } + r.closed = true + r.removed = true + r.lock.Unlock() } + rm.requests = nil rm.closed = true } +// request either creates or returns an existing mappedRequest for the given key. func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] { rm.lock.Lock() defer rm.lock.Unlock() + if rm.closed { + // return a dummy removed request for simplicity + r := &mappedRequest[K, V]{ + rm: rm, + key: key, + closed: true, + removed: true, + refCount: 1, + deliveredCh: make(chan struct{}), + } + var null V + r.deliver(null, errors.New("request map is closed")) + return r + } if r, ok := rm.requests[key]; ok { r.lock.Lock() r.refCount++ @@ -83,14 +119,6 @@ func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] { cancelFn: cancelFn, } rm.requests[key] = r - if rm.closed { - // return a closed dummy request for simplicity - r.closed = true - r.cancelFn() - var null V - r.deliver(null, errors.New("request map is closed")) - return r - } if rm.requestFn != nil { go func() { result, err := rm.requestFn(ctx, key) @@ -100,6 +128,24 @@ func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] { return r } +// remove removes the given key from the request map, ensuring that the next +// request attempt for the same key will start a new request and potentially +// yield a different result. The old mappedRequest stays accessible for those +// who had a direct reference to it. Remove does not cancel the old request if +// it is still pending. +func (rm *requestMap[K, V]) remove(key K) { + rm.lock.Lock() + defer rm.lock.Unlock() + + if r, ok := rm.requests[key]; ok { + r.lock.Lock() + r.removed = true + r.lock.Unlock() + delete(rm.requests, key) + } +} + +// has returns true if the request map has a request for the given key. func (rm *requestMap[K, V]) has(key K) bool { rm.lock.Lock() defer rm.lock.Unlock() @@ -108,6 +154,7 @@ func (rm *requestMap[K, V]) has(key K) bool { return ok } +// allKeys returns the list of keys that the map has requests associated to. func (rm *requestMap[K, V]) allKeys() []K { rm.lock.Lock() defer rm.lock.Unlock() @@ -119,7 +166,9 @@ func (rm *requestMap[K, V]) allKeys() []K { return keys } -// should only be called with validated results of successful requests +// tryDeliver delivers the given result for the request associated with the given +// key if such a request exists in the map. This function should only be called +// with a validated result. func (rm *requestMap[K, V]) tryDeliver(key K, result V) { rm.lock.Lock() defer rm.lock.Unlock() @@ -129,6 +178,7 @@ func (rm *requestMap[K, V]) tryDeliver(key K, result V) { } } +// deliver delivers either a result or an error for the request. func (r *mappedRequest[K, V]) deliver(result V, err error) { r.lock.Lock() if !r.delivered { @@ -142,7 +192,9 @@ func (r *mappedRequest[K, V]) deliver(result V, err error) { r.lock.Unlock() } -func (r *mappedRequest[K, V]) getResult(ctx context.Context) (V, error) { +// waitForResult waits for a result or an error to be delivered until the context +// is cancelled. +func (r *mappedRequest[K, V]) waitForResult(ctx context.Context) (V, error) { select { case <-r.deliveredCh: // not changed after deliveredCh is closed @@ -153,13 +205,21 @@ func (r *mappedRequest[K, V]) getResult(ctx context.Context) (V, error) { } } +// release decreases the reference counter that has been previously increased by +// request(key) and cancels and removes the request if no one is interested in it +// any more. +// Note that waitForResult does not automatically release the request so that the +// caller can store the result in the local cache before making it unavailable in +// the request map. func (r *mappedRequest[K, V]) release() { r.rm.lock.Lock() r.lock.Lock() r.refCount-- - if r.refCount == 0 { - delete(r.rm.requests, r.key) - if !r.delivered && !r.closed { + if r.refCount == 0 && !r.closed { + if !r.removed { + delete(r.rm.requests, r.key) + } + if !r.delivered { r.cancelFn() } } diff --git a/ethclient/lightclient/state.go b/ethclient/lightclient/state.go index 93eba62541..7152155e16 100644 --- a/ethclient/lightclient/state.go +++ b/ethclient/lightclient/state.go @@ -35,6 +35,7 @@ import ( "github.com/holiman/uint256" ) +// lightStateFields defines Client fields related to state proofs. type lightStateFields struct { proofCache *lru.Cache[proofRequest, *gethclient.AccountResult] proofRequests *requestMap[proofRequest, *gethclient.AccountResult] @@ -42,17 +43,22 @@ type lightStateFields struct { codeRequests *requestMap[codeRequest, []byte] } +// proofRequest defines a proof request. This structure is used as a key for the +// proof cache and the proof request map. type proofRequest struct { blockNumber uint64 address common.Address storageKeys string } +// codeRequest defines a contract code request. This structure is used as a key +// for the code cache and the code request map. type codeRequest struct { blockNumber uint64 address common.Address } +// initLightState initializes the structures related to state proofs. func (c *Client) initLightState() { c.proofCache = lru.NewCache[proofRequest, *gethclient.AccountResult](100) c.codeCache = lru.NewCache[codeRequest, []byte](10) @@ -60,11 +66,13 @@ func (c *Client) initLightState() { c.codeRequests = newRequestMap[codeRequest, []byte](c.requestCode) } +// closeLightState shuts down the structures related to state proofs. func (c *Client) closeLightState() { c.proofRequests.close() c.codeRequests.close() } +// getProof retrieves and validates a state account with an optional set of storage keys. func (c *Client) getProof(ctx context.Context, blockNumber *big.Int, account common.Address, storageKeys []string, getCode bool) (*gethclient.AccountResult, []byte, error) { proof, code, retry, err := c.getProofOnce(ctx, blockNumber, account, storageKeys, getCode) if retry { @@ -73,6 +81,13 @@ func (c *Client) getProof(ctx context.Context, blockNumber *big.Int, account com return proof, code, err } +// getProofOnce makes an attempt to retrieve and validate a state account with +// an optional set of storage keys. Note that since the queried state is selected +// by block number both in this function and the underlying RPC request and the +// client can validate the state root against the local canonical chain, it is +// possible that a request happening very close to a reorg might cause a state +// root or code hash mismatch. In this case the function clears the block number +// to proof association from the cache and returns retry == true. func (c *Client) getProofOnce(ctx context.Context, blockNumber *big.Int, account common.Address, storageKeys []string, getCode bool) (*gethclient.AccountResult, []byte, bool, error) { num, pheader, err := c.resolveBlockNumber(blockNumber) if err != nil { @@ -133,8 +148,10 @@ func (c *Client) getProofOnce(ctx context.Context, blockNumber *big.Int, account if getCode { <-codeCh c.codeCache.Remove(codeReq) + c.codeRequests.remove(codeReq) } c.proofCache.Remove(proofReq) + c.proofRequests.remove(proofReq) return nil, nil, true, err } if getCode { @@ -144,13 +161,16 @@ func (c *Client) getProofOnce(ctx context.Context, blockNumber *big.Int, account } if crypto.Keccak256Hash(code) != proof.CodeHash { c.codeCache.Remove(codeReq) + c.codeRequests.remove(codeReq) c.proofCache.Remove(proofReq) + c.proofRequests.remove(proofReq) return nil, nil, true, errors.New("code hash mismatch") } } return proof, code, false, nil } +// validateProof validates the given state proof against the specified state root. func (c *Client) validateProof(proof *gethclient.AccountResult, stateRoot common.Hash, account common.Address, storageKeys []string) error { // validate account proof proofReader, err := makeProofReader(proof.AccountProof) @@ -226,6 +246,8 @@ func (c *Client) validateProof(proof *gethclient.AccountResult, stateRoot common // proofReader implements ethdb.KeyValueReader. type proofReader map[string][]byte +// makeProofReader takes a list of hex encoded proof nodes and creates an +// ethdb.KeyValueReader that returns these proof nodes based on their hashes. func makeProofReader(proof []string) (proofReader, error) { pr := make(proofReader) for _, s := range proof { @@ -238,11 +260,13 @@ func makeProofReader(proof []string) (proofReader, error) { return pr, nil } +// Has implements ethdb.KeyValueReader. func (p proofReader) Has(key []byte) (bool, error) { _, ok := p[string(key)] return ok, nil } +// Get implements ethdb.KeyValueReader. func (p proofReader) Get(key []byte) ([]byte, error) { if value, ok := p[string(key)]; ok { return value, nil @@ -250,6 +274,7 @@ func (p proofReader) Get(key []byte) ([]byte, error) { return nil, errors.New("not found") } +// stValueBytes converts a big.Int storage value into a byte slice of 32 bytes. func stValueBytes(value *big.Int) ([]byte, error) { if value == nil { return nil, errors.New("storage value is nil") @@ -269,12 +294,17 @@ func stValueBytes(value *big.Int) ([]byte, error) { } } +// fetchProof either requests a merkle proof of a state account and optional set +// of storage keys from the RPC client or returns it from the cache if available. +// Note that proofs can only be validated in the context of the belonging canonical +// header which might be retrieved later, therefore proofs are cached before +// validation and removed later if proven invalid. func (c *Client) fetchProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) { if proof, ok := c.proofCache.Get(req); ok { return proof, nil } request := c.proofRequests.request(req) - proof, err := request.getResult(ctx) + proof, err := request.waitForResult(ctx) if err == nil { c.proofCache.Add(req, proof) // cached before validation; remove and retry if invalid } @@ -282,12 +312,17 @@ func (c *Client) fetchProof(ctx context.Context, req proofRequest) (*gethclient. return proof, err } +// fetchCode either requests the contract code at a state account from the RPC +// client or returns it from the cache if available. +// Note that proofs can only be validated in the context of the belonging canonical +// header which might be retrieved later, therefore proofs are cached before +// validation and removed later if proven invalid. func (c *Client) fetchCode(ctx context.Context, req codeRequest) ([]byte, error) { if code, ok := c.codeCache.Get(req); ok { return code, nil } request := c.codeRequests.request(req) - code, err := request.getResult(ctx) + code, err := request.waitForResult(ctx) if err == nil { c.codeCache.Add(req, code) // cached before validation; remove and retry if invalid } @@ -295,6 +330,18 @@ func (c *Client) fetchCode(ctx context.Context, req codeRequest) ([]byte, error) return code, err } +// clearStateCache removes all cached proof and code entries. +// Note that since the queried states are identified by a block number of the +// canonical chain, cached entries can be invalidated by a reorg. The canonical +// chain tracker calls this function whenever an already existing block number +// to hash association is changed. +func (c *Client) clearStateCache() { + c.proofCache.Purge() + c.codeCache.Purge() +} + +// requestProof requests a merkle proof of a state account and optional set of +// storage keys from the RPC client. func (c *Client) requestProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) { type storageResult struct { Key string `json:"key"` @@ -344,6 +391,7 @@ func (c *Client) requestProof(ctx context.Context, req proofRequest) (*gethclien return proof, err } +// requestCode requests the contract code at the given account from the RPC server. func (c *Client) requestCode(ctx context.Context, req codeRequest) ([]byte, error) { var code hexutil.Bytes log.Debug("Starting RPC request", "type", "eth_getCode", "blockNumber", req.blockNumber, "address", req.address) diff --git a/ethclient/lightclient/transactions.go b/ethclient/lightclient/transactions.go index c2124ed42e..b73e7c3d45 100644 --- a/ethclient/lightclient/transactions.go +++ b/ethclient/lightclient/transactions.go @@ -37,6 +37,7 @@ import ( const maxTxAge = 300 // sent transactions are typically remembered for an hour after last seen pending +// txAndReceiptsFields defines Client fields related to transactions and receipts. type txAndReceiptsFields struct { signer types.Signer @@ -46,45 +47,61 @@ type txAndReceiptsFields struct { txByHashRequests *requestMap[common.Hash, txResult] receiptByHashRequests *requestMap[common.Hash, *types.Receipt] - sentTxLock sync.Mutex - sentTxs map[common.Address]senderTxs + trackedTxLock sync.Mutex + trackedTxs map[common.Address]senderTxs headCounter, lastHeadNumber uint64 } +// txResult represents the results of a transaction by hash request. type txResult struct { tx *types.Transaction isPending bool } +// txInBlock stores the known inclusion positions of a transaction. +// Note that any cached position is only considered valid if the block is canonical. type txInBlock struct { blockNumber uint64 - blockHash common.Hash // only considered valid if the block is canonical + blockHash common.Hash index uint } -type sentTx struct { +// trackedTx stores the information associated with the hashes of tracked transactions. +type trackedTx struct { nonce uint64 lastSeen uint64 // headCounter value where the tx has been sent or last seen pending } -type senderTxs map[common.Hash]sentTx +// senderTxs stores all tracked transactions originating from a single sender. +type senderTxs map[common.Hash]trackedTx +// initTxAndReceipts initializes the structures related to transactions and receipts. func (c *Client) initTxAndReceipts() { c.blockReceiptsCache = lru.NewCache[common.Hash, types.Receipts](10) c.txPosCache = lru.NewCache[common.Hash, txInBlock](10000) - c.sentTxs = make(map[common.Address]senderTxs) + c.trackedTxs = make(map[common.Address]senderTxs) c.signer = types.LatestSigner(c.elConfig) c.blockReceiptsRequests = newRequestMap[common.Hash, types.Receipts](c.requestBlockReceipts) c.txByHashRequests = newRequestMap[common.Hash, txResult](c.requestTxByHash) c.receiptByHashRequests = newRequestMap[common.Hash, *types.Receipt](c.requestReceiptByTxHash) } +// closeTxAndReceipts shuts down the structures related to transactions and receipts. func (c *Client) closeTxAndReceipts() { c.blockReceiptsRequests.close() c.txByHashRequests.close() c.receiptByHashRequests.close() } +// getTxByHash requests and validates the transaction with the specified hash or +// returns it from cache if available as part of the local canonical chain. +// The pending status of the transaction is also returned. Note that the pending +// status cannot be directly validated, unless it is known to be canonical. +// Also note that if a transaction belongs to the server's latest block that is +// not proven by an optimistic update to the beacon light client yet then this +// function will report isPending == true even though the server has reported +// it as non-pending. This behavior ensures the consistency between the local +// canonical chain and the results of getTxByHash and getReceiptByTxHash. func (c *Client) getTxByHash(ctx context.Context, txHash common.Hash) (tx *types.Transaction, isPending bool, err error) { if pos, ok := c.txPosCache.Get(txHash); ok { if hash, ok := c.getCachedHash(pos.blockNumber); ok && hash == pos.blockHash { @@ -104,15 +121,21 @@ func (c *Client) getTxByHash(ctx context.Context, txHash common.Hash) (tx *types return c.getUncachedTxByHash(ctx, txHash, headBlockNumber) } +// getUncachedTxByHash requests a transaction and its pending status without +// checking in the cache whether it's already canonical. It does check for future +// inclusion and returns isPending == true if it has a receipt in a block over +// headBlockNumber. +// The purpose of this function is sharing code between getTxByHash and +// nonceAndPendingTxs. func (c *Client) getUncachedTxByHash(ctx context.Context, txHash common.Hash, headBlockNumber uint64) (tx *types.Transaction, isPending bool, err error) { req := c.txByHashRequests.request(txHash) var txResult txResult - txResult, err = req.getResult(ctx) + txResult, err = req.waitForResult(ctx) req.release() tx, isPending = txResult.tx, txResult.isPending if err == nil && !isPending { req := c.receiptByHashRequests.request(txHash) - receipt, err := req.getResult(ctx) + receipt, err := req.waitForResult(ctx) req.release() if err == ethereum.NotFound { return tx, false, nil @@ -128,6 +151,12 @@ func (c *Client) getUncachedTxByHash(ctx context.Context, txHash common.Hash, he return } +// getReceiptByTxHash requests and validates the receipt belonging to the transaction +// with the specified hash or returns it from cache if available. Results are only +// returned if they can be validated as part of the current local canonical chain. +// Note that if a receipt belongs to the server's latest block that is not proven +// by an optimistic update to the beacon light client yet then this function cannot +// validate it and will treat it like a pending transaction and return ethereum.NotFound. func (c *Client) getReceiptByTxHash(ctx context.Context, txHash common.Hash) (*types.Receipt, error) { if pos, ok := c.txPosCache.Get(txHash); ok { if hash, ok := c.getCachedHash(pos.blockNumber); ok && hash == pos.blockHash { @@ -140,7 +169,7 @@ func (c *Client) getReceiptByTxHash(ctx context.Context, txHash common.Hash) (*t } } req := c.receiptByHashRequests.request(txHash) - receipt, err := req.getResult(ctx) + receipt, err := req.waitForResult(ctx) req.release() if err != nil { return nil, err @@ -189,6 +218,8 @@ func (c *Client) getReceiptByTxHash(ctx context.Context, txHash common.Hash) (*t return receipt, err } +// getBlockReceipts requests and validates a set of block receipts or returns them +// from cache if available. func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) { if receipts, ok := c.blockReceiptsCache.Get(blockHash); ok { return receipts, nil @@ -198,7 +229,7 @@ func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (t if err != nil { return nil, err } - receipts, err := request.getResult(ctx) + receipts, err := request.waitForResult(ctx) if err == nil { receipts, err = c.validateBlockReceipts(block, receipts) } @@ -209,6 +240,13 @@ func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (t return receipts, err } +// validateBlockReceipts verifies that the given set of receipts belongs to the +// given block. Non-consensus fields of the returned set of receipts are derived +// from the block and the consensus fields of the received receipts which are also +// validated against the receipts root of the block. +// Note that the non-consensus fields of the received receipts could be ignored +// but they are also verified against the derived version in order to detect any +// potential inconsistencies between the server responses and the validation logic. func (c *Client) validateBlockReceipts(block *types.Block, receipts types.Receipts) (types.Receipts, error) { // verify consensus fields agains receipts hash in block var hash common.Hash @@ -242,11 +280,12 @@ func (c *Client) validateBlockReceipts(block *types.Block, receipts types.Receip return nil, err } // compare the JSON encoding of received and derived versions + // Note: this step could be skipped but it is performed as a sanity check. jsonReceived, err := json.Marshal(receipts) if err != nil { return nil, err } - jsonDerived, err := json.Marshal(receipts) + jsonDerived, err := json.Marshal(newReceipts) if err != nil { return nil, err } @@ -256,6 +295,11 @@ func (c *Client) validateBlockReceipts(block *types.Block, receipts types.Receip return newReceipts, nil } +// requestTxByHash requests the transaction belonging to the specified hash from +// the RPC client, along with the current pending status. +// Either a transaction with the specified hash or an error is returned. Note that +// the pending status cannot be directly validated because the light client does +// not track the set of pending transactions. func (c *Client) requestTxByHash(ctx context.Context, hash common.Hash) (txResult, error) { var json *rpcTransaction err := c.client.CallContext(ctx, &json, "eth_getTransactionByHash", hash) @@ -274,6 +318,12 @@ func (c *Client) requestTxByHash(ctx context.Context, hash common.Hash) (txResul return txResult{tx: json.tx, isPending: json.BlockNumber == nil}, nil } +// requestReceiptByTxHash requests the receipt belonging to the transaction with +// the specified hash from the RPC client. +// Either a receipt or an error is returned. Note that the results are not validated +// at this level as it also requires the block in which the transaction is included. +// Also note that since the receipt depends on the inclusion block, the hash to +// receipt association also depends on the current canonical chain. func (c *Client) requestReceiptByTxHash(ctx context.Context, txHash common.Hash) (*types.Receipt, error) { var r *types.Receipt err := c.client.CallContext(ctx, &r, "eth_getTransactionReceipt", txHash) @@ -283,6 +333,10 @@ func (c *Client) requestReceiptByTxHash(ctx context.Context, txHash common.Hash) return r, err } +// requestBlockReceipts requests the set of receipts belonging to the block with +// the specified hash from the RPC client. +// Either a set of receipts or an error is returned. Note that the results are not +// validated at this level as it also requires the belonging block. func (c *Client) requestBlockReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) { var r []*types.Receipt err := c.client.CallContext(ctx, &r, "eth_getBlockReceipts", blockHash) @@ -292,6 +346,9 @@ func (c *Client) requestBlockReceipts(ctx context.Context, blockHash common.Hash return types.Receipts(r), err } +// cacheBlockTxPositions iterates through the transactions of a block and caches +// transaction position information. This function should be called for all +// new canonical blocks. func (c *Client) cacheBlockTxPositions(block *types.Block) { blockNumber, blockHash := block.NumberU64(), block.Hash() for i, tx := range block.Transactions() { @@ -299,6 +356,8 @@ func (c *Client) cacheBlockTxPositions(block *types.Block) { } } +// sendTransaction sends a transaction to the RPC server and adds it to the set of +// tracked transactions. func (c *Client) sendTransaction(ctx context.Context, tx *types.Transaction) error { sender, err := types.Sender(c.signer, tx) if err != nil { @@ -311,12 +370,15 @@ func (c *Client) sendTransaction(ctx context.Context, tx *types.Transaction) err if err := c.client.CallContext(ctx, nil, "eth_sendRawTransaction", hexutil.Encode(data)); err != nil { return err } - c.sentTxLock.Lock() + c.trackedTxLock.Lock() c.markTxAsSeen(sender, tx) - c.sentTxLock.Unlock() + c.trackedTxLock.Unlock() return nil } +// nonceAndPendingTxs obtains the latest nonce of the given sender account and checks +// the pending status of the tracked transactions originating from that server. +// It returns the current nonce and a nonce-ordered list of pending transactions. func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionHeader, sender common.Address) (uint64, types.Transactions, error) { proof, err := c.fetchProof(ctx, proofRequest{blockNumber: head.BlockNumber(), address: sender, storageKeys: ""}) if err != nil { @@ -326,9 +388,9 @@ func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionH return 0, nil, err } - c.sentTxLock.Lock() - senderTxs := c.sentTxs[sender] - c.sentTxLock.Unlock() + c.trackedTxLock.Lock() + senderTxs := c.trackedTxs[sender] + c.trackedTxLock.Unlock() type pendingResult struct { pendingTx *types.Transaction @@ -337,8 +399,8 @@ func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionH resultCh := make(chan pendingResult, len(senderTxs)) var reqCount int - for txHash, sentTx := range senderTxs { - if sentTx.nonce <= proof.Nonce { + for txHash, trackedTx := range senderTxs { + if trackedTx.nonce <= proof.Nonce { continue } reqCount++ @@ -362,52 +424,59 @@ func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionH } } sort.Sort(types.TxByNonce(pendingList)) - c.sentTxLock.Lock() + c.trackedTxLock.Lock() for _, tx := range pendingList { c.markTxAsSeen(sender, tx) } - c.sentTxLock.Unlock() + c.trackedTxLock.Unlock() return proof.Nonce, pendingList, nil } +// allSenders returns the list of all senders that have belonging tracked transactions. func (c *Client) allSenders() []common.Address { - c.sentTxLock.Lock() - allSenders := make([]common.Address, 0, len(c.sentTxs)) - for sender := range c.sentTxs { + c.trackedTxLock.Lock() + allSenders := make([]common.Address, 0, len(c.trackedTxs)) + for sender := range c.trackedTxs { allSenders = append(allSenders, sender) } - c.sentTxLock.Unlock() + c.trackedTxLock.Unlock() return allSenders } +// txAndReceiptsNewHead should be called for every new head. It counts the head +// updates that increase the current block height and discards transactions that +// have not been seen as pending lately based on this head counter. func (c *Client) txAndReceiptsNewHead(number uint64, hash common.Hash) { - c.sentTxLock.Lock() + c.trackedTxLock.Lock() if number > c.lastHeadNumber { c.lastHeadNumber = number c.headCounter++ c.discardOldTxs() } - c.sentTxLock.Unlock() + c.trackedTxLock.Unlock() } +// discardOldTxs discards transactions that have not been seen as pending lately. func (c *Client) discardOldTxs() { - for sender, senderTxs := range c.sentTxs { - for txHash, sentTx := range senderTxs { - if sentTx.lastSeen+maxTxAge < c.headCounter { + for sender, senderTxs := range c.trackedTxs { + for txHash, trackedTx := range senderTxs { + if trackedTx.lastSeen+maxTxAge < c.headCounter { delete(senderTxs, txHash) } } if len(senderTxs) == 0 { - delete(c.sentTxs, sender) + delete(c.trackedTxs, sender) } } } +// markTxAsSeen adds a transaction to the tracked set or updates its last seen +// counter in order to keep it in the set. func (c *Client) markTxAsSeen(sender common.Address, tx *types.Transaction) { - senderTxs := c.sentTxs[sender] + senderTxs := c.trackedTxs[sender] if senderTxs == nil { - senderTxs = make(map[common.Hash]sentTx) - c.sentTxs[sender] = senderTxs + senderTxs = make(map[common.Hash]trackedTx) + c.trackedTxs[sender] = senderTxs } - senderTxs[tx.Hash()] = sentTx{nonce: tx.Nonce(), lastSeen: c.headCounter} + senderTxs[tx.Hash()] = trackedTx{nonce: tx.Nonce(), lastSeen: c.headCounter} }