mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 18:32:23 +00:00
ethclient/lightclient: added comments and minor fixes
This commit is contained in:
parent
40e1577e67
commit
84f1e31909
5 changed files with 254 additions and 66 deletions
|
|
@ -30,6 +30,7 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// blocksAndHeadersFields defines Client fields related to blocks and headers.
|
||||||
type blocksAndHeadersFields struct {
|
type blocksAndHeadersFields struct {
|
||||||
headerCache *lru.Cache[common.Hash, *types.Header]
|
headerCache *lru.Cache[common.Hash, *types.Header]
|
||||||
headerRequests *requestMap[common.Hash, *types.Header]
|
headerRequests *requestMap[common.Hash, *types.Header]
|
||||||
|
|
@ -38,6 +39,7 @@ type blocksAndHeadersFields struct {
|
||||||
blockRequests *requestMap[common.Hash, *types.Block]
|
blockRequests *requestMap[common.Hash, *types.Block]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initBlocksAndHeaders initializes the structures related to blocks and headers.
|
||||||
func (c *Client) initBlocksAndHeaders() {
|
func (c *Client) initBlocksAndHeaders() {
|
||||||
c.headerCache = lru.NewCache[common.Hash, *types.Header](1000)
|
c.headerCache = lru.NewCache[common.Hash, *types.Header](1000)
|
||||||
c.payloadHeaderCache = lru.NewCache[common.Hash, *btypes.ExecutionHeader](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)
|
c.blockRequests = newRequestMap[common.Hash, *types.Block](c.requestBlock)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// closeBlocksAndHeaders shuts down the structures related to blocks and headers.
|
||||||
func (c *Client) closeBlocksAndHeaders() {
|
func (c *Client) closeBlocksAndHeaders() {
|
||||||
c.headerRequests.close()
|
c.headerRequests.close()
|
||||||
c.blockRequests.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) {
|
if c.blockRequests.has(hash) && !c.headerRequests.has(hash) {
|
||||||
req := c.blockRequests.request(hash)
|
req := c.blockRequests.request(hash)
|
||||||
block, err := req.getResult(ctx)
|
block, err := req.waitForResult(ctx)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
header := block.Header()
|
header := block.Header()
|
||||||
c.headerCache.Add(hash, 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)
|
req := c.headerRequests.request(hash)
|
||||||
header, err := req.getResult(ctx)
|
header, err := req.waitForResult(ctx)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
c.headerCache.Add(hash, header)
|
c.headerCache.Add(hash, header)
|
||||||
}
|
}
|
||||||
|
|
@ -95,7 +98,7 @@ func (c *Client) getBlock(ctx context.Context, hash common.Hash) (*types.Block,
|
||||||
return block, nil
|
return block, nil
|
||||||
}
|
}
|
||||||
req := c.blockRequests.request(hash)
|
req := c.blockRequests.request(hash)
|
||||||
block, err := req.getResult(ctx)
|
block, err := req.waitForResult(ctx)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
header := block.Header()
|
header := block.Header()
|
||||||
c.headerCache.Add(hash, 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
|
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) {
|
func (c *Client) requestHeader(ctx context.Context, hash common.Hash) (*types.Header, error) {
|
||||||
var header *types.Header
|
var header *types.Header
|
||||||
log.Debug("Starting RPC request", "type", "eth_getBlockByHash", "hash", hash, "full", false)
|
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
|
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) {
|
func (c *Client) requestBlock(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||||
var (
|
var (
|
||||||
raw json.RawMessage
|
raw json.RawMessage
|
||||||
|
|
|
||||||
|
|
@ -33,6 +33,7 @@ import (
|
||||||
|
|
||||||
const recentCanonicalLength = 256
|
const recentCanonicalLength = 256
|
||||||
|
|
||||||
|
// canonicalChainFields defines Client fields related to canonical chain tracking.
|
||||||
type canonicalChainFields struct {
|
type canonicalChainFields struct {
|
||||||
chainLock sync.Mutex
|
chainLock sync.Mutex
|
||||||
head, finality *btypes.ExecutionHeader
|
head, finality *btypes.ExecutionHeader
|
||||||
|
|
@ -43,6 +44,7 @@ type canonicalChainFields struct {
|
||||||
requests *requestMap[uint64, common.Hash] // requested; neither recent nor cached finalized
|
requests *requestMap[uint64, common.Hash] // requested; neither recent nor cached finalized
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initCanonicalChain initializes the structures related to canonical chain tracking.
|
||||||
func (c *Client) initCanonicalChain() {
|
func (c *Client) initCanonicalChain() {
|
||||||
c.finalized = lru.NewCache[uint64, common.Hash](10000)
|
c.finalized = lru.NewCache[uint64, common.Hash](10000)
|
||||||
c.requests = newRequestMap[uint64, common.Hash](nil)
|
c.requests = newRequestMap[uint64, common.Hash](nil)
|
||||||
|
|
@ -51,6 +53,7 @@ func (c *Client) initCanonicalChain() {
|
||||||
go c.tailFetcher()
|
go c.tailFetcher()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// closeCanonicalChain shuts down the structures related to canonical chain tracking.
|
||||||
func (c *Client) closeCanonicalChain() {
|
func (c *Client) closeCanonicalChain() {
|
||||||
c.requests.close()
|
c.requests.close()
|
||||||
close(c.closeCh)
|
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() {
|
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
|
// initialize recent canonical hash map when first head is added or when
|
||||||
// it is not a descendant of the previous head
|
// it is not a descendant of the previous head
|
||||||
|
c.clearStateCache()
|
||||||
c.recent = make(map[uint64]common.Hash)
|
c.recent = make(map[uint64]common.Hash)
|
||||||
if headNum > 0 {
|
if headNum > 0 {
|
||||||
c.recent[headNum-1] = head.ParentHash()
|
c.recent[headNum-1] = head.ParentHash()
|
||||||
|
|
@ -179,7 +183,7 @@ func (c *Client) getHash(ctx context.Context, number uint64) (common.Hash, error
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
defer req.release()
|
defer req.release()
|
||||||
return req.getResult(ctx)
|
return req.waitForResult(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Client) getCachedHash(number uint64) (common.Hash, bool) {
|
func (c *Client) getCachedHash(number uint64) (common.Hash, bool) {
|
||||||
|
|
|
||||||
|
|
@ -22,6 +22,11 @@ import (
|
||||||
"sync"
|
"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 {
|
type requestMap[K comparable, V any] struct {
|
||||||
lock sync.Mutex
|
lock sync.Mutex
|
||||||
closed bool
|
closed bool
|
||||||
|
|
@ -29,18 +34,25 @@ type requestMap[K comparable, V any] struct {
|
||||||
requests map[K]*mappedRequest[K, V]
|
requests map[K]*mappedRequest[K, V]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// mappedRequest represents a request that multiple processes can join.
|
||||||
type mappedRequest[K comparable, V any] struct {
|
type mappedRequest[K comparable, V any] struct {
|
||||||
lock sync.Mutex
|
lock sync.Mutex
|
||||||
rm *requestMap[K, V]
|
rm *requestMap[K, V] // in the map if !closed && !removed && refCount != 0
|
||||||
key K
|
key K
|
||||||
refCount int
|
refCount int
|
||||||
delivered, closed bool
|
closed, delivered, removed bool
|
||||||
deliveredCh chan struct{}
|
deliveredCh chan struct{}
|
||||||
cancelFn func() // called when delivered || closed || refCount == 0 becomes true
|
cancelFn func() // called when closed || delivered || refCount == 0 becomes true
|
||||||
result V
|
result V
|
||||||
err error
|
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] {
|
func newRequestMap[K comparable, V any](requestFn func(context.Context, K) (V, error)) *requestMap[K, V] {
|
||||||
return &requestMap[K, V]{
|
return &requestMap[K, V]{
|
||||||
requestFn: requestFn,
|
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() {
|
func (rm *requestMap[K, V]) close() {
|
||||||
rm.lock.Lock()
|
rm.lock.Lock()
|
||||||
defer rm.lock.Unlock()
|
defer rm.lock.Unlock()
|
||||||
|
|
@ -55,19 +72,38 @@ func (rm *requestMap[K, V]) close() {
|
||||||
if rm.closed {
|
if rm.closed {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for _, req := range rm.requests {
|
for _, r := range rm.requests {
|
||||||
if !req.delivered && req.refCount != 0 {
|
r.lock.Lock()
|
||||||
req.cancelFn()
|
if !r.delivered && r.refCount != 0 {
|
||||||
req.closed = true
|
r.cancelFn()
|
||||||
}
|
}
|
||||||
|
r.closed = true
|
||||||
|
r.removed = true
|
||||||
|
r.lock.Unlock()
|
||||||
}
|
}
|
||||||
|
rm.requests = nil
|
||||||
rm.closed = true
|
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] {
|
func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] {
|
||||||
rm.lock.Lock()
|
rm.lock.Lock()
|
||||||
defer rm.lock.Unlock()
|
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 {
|
if r, ok := rm.requests[key]; ok {
|
||||||
r.lock.Lock()
|
r.lock.Lock()
|
||||||
r.refCount++
|
r.refCount++
|
||||||
|
|
@ -83,14 +119,6 @@ func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] {
|
||||||
cancelFn: cancelFn,
|
cancelFn: cancelFn,
|
||||||
}
|
}
|
||||||
rm.requests[key] = r
|
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 {
|
if rm.requestFn != nil {
|
||||||
go func() {
|
go func() {
|
||||||
result, err := rm.requestFn(ctx, key)
|
result, err := rm.requestFn(ctx, key)
|
||||||
|
|
@ -100,6 +128,24 @@ func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] {
|
||||||
return r
|
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 {
|
func (rm *requestMap[K, V]) has(key K) bool {
|
||||||
rm.lock.Lock()
|
rm.lock.Lock()
|
||||||
defer rm.lock.Unlock()
|
defer rm.lock.Unlock()
|
||||||
|
|
@ -108,6 +154,7 @@ func (rm *requestMap[K, V]) has(key K) bool {
|
||||||
return ok
|
return ok
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// allKeys returns the list of keys that the map has requests associated to.
|
||||||
func (rm *requestMap[K, V]) allKeys() []K {
|
func (rm *requestMap[K, V]) allKeys() []K {
|
||||||
rm.lock.Lock()
|
rm.lock.Lock()
|
||||||
defer rm.lock.Unlock()
|
defer rm.lock.Unlock()
|
||||||
|
|
@ -119,7 +166,9 @@ func (rm *requestMap[K, V]) allKeys() []K {
|
||||||
return keys
|
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) {
|
func (rm *requestMap[K, V]) tryDeliver(key K, result V) {
|
||||||
rm.lock.Lock()
|
rm.lock.Lock()
|
||||||
defer rm.lock.Unlock()
|
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) {
|
func (r *mappedRequest[K, V]) deliver(result V, err error) {
|
||||||
r.lock.Lock()
|
r.lock.Lock()
|
||||||
if !r.delivered {
|
if !r.delivered {
|
||||||
|
|
@ -142,7 +192,9 @@ func (r *mappedRequest[K, V]) deliver(result V, err error) {
|
||||||
r.lock.Unlock()
|
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 {
|
select {
|
||||||
case <-r.deliveredCh:
|
case <-r.deliveredCh:
|
||||||
// not changed after deliveredCh is closed
|
// 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() {
|
func (r *mappedRequest[K, V]) release() {
|
||||||
r.rm.lock.Lock()
|
r.rm.lock.Lock()
|
||||||
r.lock.Lock()
|
r.lock.Lock()
|
||||||
r.refCount--
|
r.refCount--
|
||||||
if r.refCount == 0 {
|
if r.refCount == 0 && !r.closed {
|
||||||
|
if !r.removed {
|
||||||
delete(r.rm.requests, r.key)
|
delete(r.rm.requests, r.key)
|
||||||
if !r.delivered && !r.closed {
|
}
|
||||||
|
if !r.delivered {
|
||||||
r.cancelFn()
|
r.cancelFn()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -35,6 +35,7 @@ import (
|
||||||
"github.com/holiman/uint256"
|
"github.com/holiman/uint256"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// lightStateFields defines Client fields related to state proofs.
|
||||||
type lightStateFields struct {
|
type lightStateFields struct {
|
||||||
proofCache *lru.Cache[proofRequest, *gethclient.AccountResult]
|
proofCache *lru.Cache[proofRequest, *gethclient.AccountResult]
|
||||||
proofRequests *requestMap[proofRequest, *gethclient.AccountResult]
|
proofRequests *requestMap[proofRequest, *gethclient.AccountResult]
|
||||||
|
|
@ -42,17 +43,22 @@ type lightStateFields struct {
|
||||||
codeRequests *requestMap[codeRequest, []byte]
|
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 {
|
type proofRequest struct {
|
||||||
blockNumber uint64
|
blockNumber uint64
|
||||||
address common.Address
|
address common.Address
|
||||||
storageKeys string
|
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 {
|
type codeRequest struct {
|
||||||
blockNumber uint64
|
blockNumber uint64
|
||||||
address common.Address
|
address common.Address
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initLightState initializes the structures related to state proofs.
|
||||||
func (c *Client) initLightState() {
|
func (c *Client) initLightState() {
|
||||||
c.proofCache = lru.NewCache[proofRequest, *gethclient.AccountResult](100)
|
c.proofCache = lru.NewCache[proofRequest, *gethclient.AccountResult](100)
|
||||||
c.codeCache = lru.NewCache[codeRequest, []byte](10)
|
c.codeCache = lru.NewCache[codeRequest, []byte](10)
|
||||||
|
|
@ -60,11 +66,13 @@ func (c *Client) initLightState() {
|
||||||
c.codeRequests = newRequestMap[codeRequest, []byte](c.requestCode)
|
c.codeRequests = newRequestMap[codeRequest, []byte](c.requestCode)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// closeLightState shuts down the structures related to state proofs.
|
||||||
func (c *Client) closeLightState() {
|
func (c *Client) closeLightState() {
|
||||||
c.proofRequests.close()
|
c.proofRequests.close()
|
||||||
c.codeRequests.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) {
|
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)
|
proof, code, retry, err := c.getProofOnce(ctx, blockNumber, account, storageKeys, getCode)
|
||||||
if retry {
|
if retry {
|
||||||
|
|
@ -73,6 +81,13 @@ func (c *Client) getProof(ctx context.Context, blockNumber *big.Int, account com
|
||||||
return proof, code, err
|
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) {
|
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)
|
num, pheader, err := c.resolveBlockNumber(blockNumber)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -133,8 +148,10 @@ func (c *Client) getProofOnce(ctx context.Context, blockNumber *big.Int, account
|
||||||
if getCode {
|
if getCode {
|
||||||
<-codeCh
|
<-codeCh
|
||||||
c.codeCache.Remove(codeReq)
|
c.codeCache.Remove(codeReq)
|
||||||
|
c.codeRequests.remove(codeReq)
|
||||||
}
|
}
|
||||||
c.proofCache.Remove(proofReq)
|
c.proofCache.Remove(proofReq)
|
||||||
|
c.proofRequests.remove(proofReq)
|
||||||
return nil, nil, true, err
|
return nil, nil, true, err
|
||||||
}
|
}
|
||||||
if getCode {
|
if getCode {
|
||||||
|
|
@ -144,13 +161,16 @@ func (c *Client) getProofOnce(ctx context.Context, blockNumber *big.Int, account
|
||||||
}
|
}
|
||||||
if crypto.Keccak256Hash(code) != proof.CodeHash {
|
if crypto.Keccak256Hash(code) != proof.CodeHash {
|
||||||
c.codeCache.Remove(codeReq)
|
c.codeCache.Remove(codeReq)
|
||||||
|
c.codeRequests.remove(codeReq)
|
||||||
c.proofCache.Remove(proofReq)
|
c.proofCache.Remove(proofReq)
|
||||||
|
c.proofRequests.remove(proofReq)
|
||||||
return nil, nil, true, errors.New("code hash mismatch")
|
return nil, nil, true, errors.New("code hash mismatch")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return proof, code, false, nil
|
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 {
|
func (c *Client) validateProof(proof *gethclient.AccountResult, stateRoot common.Hash, account common.Address, storageKeys []string) error {
|
||||||
// validate account proof
|
// validate account proof
|
||||||
proofReader, err := makeProofReader(proof.AccountProof)
|
proofReader, err := makeProofReader(proof.AccountProof)
|
||||||
|
|
@ -226,6 +246,8 @@ func (c *Client) validateProof(proof *gethclient.AccountResult, stateRoot common
|
||||||
// proofReader implements ethdb.KeyValueReader.
|
// proofReader implements ethdb.KeyValueReader.
|
||||||
type proofReader map[string][]byte
|
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) {
|
func makeProofReader(proof []string) (proofReader, error) {
|
||||||
pr := make(proofReader)
|
pr := make(proofReader)
|
||||||
for _, s := range proof {
|
for _, s := range proof {
|
||||||
|
|
@ -238,11 +260,13 @@ func makeProofReader(proof []string) (proofReader, error) {
|
||||||
return pr, nil
|
return pr, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Has implements ethdb.KeyValueReader.
|
||||||
func (p proofReader) Has(key []byte) (bool, error) {
|
func (p proofReader) Has(key []byte) (bool, error) {
|
||||||
_, ok := p[string(key)]
|
_, ok := p[string(key)]
|
||||||
return ok, nil
|
return ok, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Get implements ethdb.KeyValueReader.
|
||||||
func (p proofReader) Get(key []byte) ([]byte, error) {
|
func (p proofReader) Get(key []byte) ([]byte, error) {
|
||||||
if value, ok := p[string(key)]; ok {
|
if value, ok := p[string(key)]; ok {
|
||||||
return value, nil
|
return value, nil
|
||||||
|
|
@ -250,6 +274,7 @@ func (p proofReader) Get(key []byte) ([]byte, error) {
|
||||||
return nil, errors.New("not found")
|
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) {
|
func stValueBytes(value *big.Int) ([]byte, error) {
|
||||||
if value == nil {
|
if value == nil {
|
||||||
return nil, errors.New("storage value is 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) {
|
func (c *Client) fetchProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) {
|
||||||
if proof, ok := c.proofCache.Get(req); ok {
|
if proof, ok := c.proofCache.Get(req); ok {
|
||||||
return proof, nil
|
return proof, nil
|
||||||
}
|
}
|
||||||
request := c.proofRequests.request(req)
|
request := c.proofRequests.request(req)
|
||||||
proof, err := request.getResult(ctx)
|
proof, err := request.waitForResult(ctx)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
c.proofCache.Add(req, proof) // cached before validation; remove and retry if invalid
|
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
|
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) {
|
func (c *Client) fetchCode(ctx context.Context, req codeRequest) ([]byte, error) {
|
||||||
if code, ok := c.codeCache.Get(req); ok {
|
if code, ok := c.codeCache.Get(req); ok {
|
||||||
return code, nil
|
return code, nil
|
||||||
}
|
}
|
||||||
request := c.codeRequests.request(req)
|
request := c.codeRequests.request(req)
|
||||||
code, err := request.getResult(ctx)
|
code, err := request.waitForResult(ctx)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
c.codeCache.Add(req, code) // cached before validation; remove and retry if invalid
|
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
|
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) {
|
func (c *Client) requestProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) {
|
||||||
type storageResult struct {
|
type storageResult struct {
|
||||||
Key string `json:"key"`
|
Key string `json:"key"`
|
||||||
|
|
@ -344,6 +391,7 @@ func (c *Client) requestProof(ctx context.Context, req proofRequest) (*gethclien
|
||||||
return proof, err
|
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) {
|
func (c *Client) requestCode(ctx context.Context, req codeRequest) ([]byte, error) {
|
||||||
var code hexutil.Bytes
|
var code hexutil.Bytes
|
||||||
log.Debug("Starting RPC request", "type", "eth_getCode", "blockNumber", req.blockNumber, "address", req.address)
|
log.Debug("Starting RPC request", "type", "eth_getCode", "blockNumber", req.blockNumber, "address", req.address)
|
||||||
|
|
|
||||||
|
|
@ -37,6 +37,7 @@ import (
|
||||||
|
|
||||||
const maxTxAge = 300 // sent transactions are typically remembered for an hour after last seen pending
|
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 {
|
type txAndReceiptsFields struct {
|
||||||
signer types.Signer
|
signer types.Signer
|
||||||
|
|
||||||
|
|
@ -46,45 +47,61 @@ type txAndReceiptsFields struct {
|
||||||
txByHashRequests *requestMap[common.Hash, txResult]
|
txByHashRequests *requestMap[common.Hash, txResult]
|
||||||
receiptByHashRequests *requestMap[common.Hash, *types.Receipt]
|
receiptByHashRequests *requestMap[common.Hash, *types.Receipt]
|
||||||
|
|
||||||
sentTxLock sync.Mutex
|
trackedTxLock sync.Mutex
|
||||||
sentTxs map[common.Address]senderTxs
|
trackedTxs map[common.Address]senderTxs
|
||||||
headCounter, lastHeadNumber uint64
|
headCounter, lastHeadNumber uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// txResult represents the results of a transaction by hash request.
|
||||||
type txResult struct {
|
type txResult struct {
|
||||||
tx *types.Transaction
|
tx *types.Transaction
|
||||||
isPending bool
|
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 {
|
type txInBlock struct {
|
||||||
blockNumber uint64
|
blockNumber uint64
|
||||||
blockHash common.Hash // only considered valid if the block is canonical
|
blockHash common.Hash
|
||||||
index uint
|
index uint
|
||||||
}
|
}
|
||||||
|
|
||||||
type sentTx struct {
|
// trackedTx stores the information associated with the hashes of tracked transactions.
|
||||||
|
type trackedTx struct {
|
||||||
nonce uint64
|
nonce uint64
|
||||||
lastSeen uint64 // headCounter value where the tx has been sent or last seen pending
|
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() {
|
func (c *Client) initTxAndReceipts() {
|
||||||
c.blockReceiptsCache = lru.NewCache[common.Hash, types.Receipts](10)
|
c.blockReceiptsCache = lru.NewCache[common.Hash, types.Receipts](10)
|
||||||
c.txPosCache = lru.NewCache[common.Hash, txInBlock](10000)
|
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.signer = types.LatestSigner(c.elConfig)
|
||||||
c.blockReceiptsRequests = newRequestMap[common.Hash, types.Receipts](c.requestBlockReceipts)
|
c.blockReceiptsRequests = newRequestMap[common.Hash, types.Receipts](c.requestBlockReceipts)
|
||||||
c.txByHashRequests = newRequestMap[common.Hash, txResult](c.requestTxByHash)
|
c.txByHashRequests = newRequestMap[common.Hash, txResult](c.requestTxByHash)
|
||||||
c.receiptByHashRequests = newRequestMap[common.Hash, *types.Receipt](c.requestReceiptByTxHash)
|
c.receiptByHashRequests = newRequestMap[common.Hash, *types.Receipt](c.requestReceiptByTxHash)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// closeTxAndReceipts shuts down the structures related to transactions and receipts.
|
||||||
func (c *Client) closeTxAndReceipts() {
|
func (c *Client) closeTxAndReceipts() {
|
||||||
c.blockReceiptsRequests.close()
|
c.blockReceiptsRequests.close()
|
||||||
c.txByHashRequests.close()
|
c.txByHashRequests.close()
|
||||||
c.receiptByHashRequests.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) {
|
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 pos, ok := c.txPosCache.Get(txHash); ok {
|
||||||
if hash, ok := c.getCachedHash(pos.blockNumber); ok && hash == pos.blockHash {
|
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)
|
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) {
|
func (c *Client) getUncachedTxByHash(ctx context.Context, txHash common.Hash, headBlockNumber uint64) (tx *types.Transaction, isPending bool, err error) {
|
||||||
req := c.txByHashRequests.request(txHash)
|
req := c.txByHashRequests.request(txHash)
|
||||||
var txResult txResult
|
var txResult txResult
|
||||||
txResult, err = req.getResult(ctx)
|
txResult, err = req.waitForResult(ctx)
|
||||||
req.release()
|
req.release()
|
||||||
tx, isPending = txResult.tx, txResult.isPending
|
tx, isPending = txResult.tx, txResult.isPending
|
||||||
if err == nil && !isPending {
|
if err == nil && !isPending {
|
||||||
req := c.receiptByHashRequests.request(txHash)
|
req := c.receiptByHashRequests.request(txHash)
|
||||||
receipt, err := req.getResult(ctx)
|
receipt, err := req.waitForResult(ctx)
|
||||||
req.release()
|
req.release()
|
||||||
if err == ethereum.NotFound {
|
if err == ethereum.NotFound {
|
||||||
return tx, false, nil
|
return tx, false, nil
|
||||||
|
|
@ -128,6 +151,12 @@ func (c *Client) getUncachedTxByHash(ctx context.Context, txHash common.Hash, he
|
||||||
return
|
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) {
|
func (c *Client) getReceiptByTxHash(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
|
||||||
if pos, ok := c.txPosCache.Get(txHash); ok {
|
if pos, ok := c.txPosCache.Get(txHash); ok {
|
||||||
if hash, ok := c.getCachedHash(pos.blockNumber); ok && hash == pos.blockHash {
|
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)
|
req := c.receiptByHashRequests.request(txHash)
|
||||||
receipt, err := req.getResult(ctx)
|
receipt, err := req.waitForResult(ctx)
|
||||||
req.release()
|
req.release()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
@ -189,6 +218,8 @@ func (c *Client) getReceiptByTxHash(ctx context.Context, txHash common.Hash) (*t
|
||||||
return receipt, err
|
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) {
|
func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) {
|
||||||
if receipts, ok := c.blockReceiptsCache.Get(blockHash); ok {
|
if receipts, ok := c.blockReceiptsCache.Get(blockHash); ok {
|
||||||
return receipts, nil
|
return receipts, nil
|
||||||
|
|
@ -198,7 +229,7 @@ func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (t
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
receipts, err := request.getResult(ctx)
|
receipts, err := request.waitForResult(ctx)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
receipts, err = c.validateBlockReceipts(block, receipts)
|
receipts, err = c.validateBlockReceipts(block, receipts)
|
||||||
}
|
}
|
||||||
|
|
@ -209,6 +240,13 @@ func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (t
|
||||||
return receipts, err
|
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) {
|
func (c *Client) validateBlockReceipts(block *types.Block, receipts types.Receipts) (types.Receipts, error) {
|
||||||
// verify consensus fields agains receipts hash in block
|
// verify consensus fields agains receipts hash in block
|
||||||
var hash common.Hash
|
var hash common.Hash
|
||||||
|
|
@ -242,11 +280,12 @@ func (c *Client) validateBlockReceipts(block *types.Block, receipts types.Receip
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
// compare the JSON encoding of received and derived versions
|
// 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)
|
jsonReceived, err := json.Marshal(receipts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
jsonDerived, err := json.Marshal(receipts)
|
jsonDerived, err := json.Marshal(newReceipts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
@ -256,6 +295,11 @@ func (c *Client) validateBlockReceipts(block *types.Block, receipts types.Receip
|
||||||
return newReceipts, nil
|
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) {
|
func (c *Client) requestTxByHash(ctx context.Context, hash common.Hash) (txResult, error) {
|
||||||
var json *rpcTransaction
|
var json *rpcTransaction
|
||||||
err := c.client.CallContext(ctx, &json, "eth_getTransactionByHash", hash)
|
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
|
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) {
|
func (c *Client) requestReceiptByTxHash(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
|
||||||
var r *types.Receipt
|
var r *types.Receipt
|
||||||
err := c.client.CallContext(ctx, &r, "eth_getTransactionReceipt", txHash)
|
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
|
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) {
|
func (c *Client) requestBlockReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) {
|
||||||
var r []*types.Receipt
|
var r []*types.Receipt
|
||||||
err := c.client.CallContext(ctx, &r, "eth_getBlockReceipts", blockHash)
|
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
|
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) {
|
func (c *Client) cacheBlockTxPositions(block *types.Block) {
|
||||||
blockNumber, blockHash := block.NumberU64(), block.Hash()
|
blockNumber, blockHash := block.NumberU64(), block.Hash()
|
||||||
for i, tx := range block.Transactions() {
|
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 {
|
func (c *Client) sendTransaction(ctx context.Context, tx *types.Transaction) error {
|
||||||
sender, err := types.Sender(c.signer, tx)
|
sender, err := types.Sender(c.signer, tx)
|
||||||
if err != nil {
|
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 {
|
if err := c.client.CallContext(ctx, nil, "eth_sendRawTransaction", hexutil.Encode(data)); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
c.sentTxLock.Lock()
|
c.trackedTxLock.Lock()
|
||||||
c.markTxAsSeen(sender, tx)
|
c.markTxAsSeen(sender, tx)
|
||||||
c.sentTxLock.Unlock()
|
c.trackedTxLock.Unlock()
|
||||||
return nil
|
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) {
|
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: ""})
|
proof, err := c.fetchProof(ctx, proofRequest{blockNumber: head.BlockNumber(), address: sender, storageKeys: ""})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -326,9 +388,9 @@ func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionH
|
||||||
return 0, nil, err
|
return 0, nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
c.sentTxLock.Lock()
|
c.trackedTxLock.Lock()
|
||||||
senderTxs := c.sentTxs[sender]
|
senderTxs := c.trackedTxs[sender]
|
||||||
c.sentTxLock.Unlock()
|
c.trackedTxLock.Unlock()
|
||||||
|
|
||||||
type pendingResult struct {
|
type pendingResult struct {
|
||||||
pendingTx *types.Transaction
|
pendingTx *types.Transaction
|
||||||
|
|
@ -337,8 +399,8 @@ func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionH
|
||||||
|
|
||||||
resultCh := make(chan pendingResult, len(senderTxs))
|
resultCh := make(chan pendingResult, len(senderTxs))
|
||||||
var reqCount int
|
var reqCount int
|
||||||
for txHash, sentTx := range senderTxs {
|
for txHash, trackedTx := range senderTxs {
|
||||||
if sentTx.nonce <= proof.Nonce {
|
if trackedTx.nonce <= proof.Nonce {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
reqCount++
|
reqCount++
|
||||||
|
|
@ -362,52 +424,59 @@ func (c *Client) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionH
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
sort.Sort(types.TxByNonce(pendingList))
|
sort.Sort(types.TxByNonce(pendingList))
|
||||||
c.sentTxLock.Lock()
|
c.trackedTxLock.Lock()
|
||||||
for _, tx := range pendingList {
|
for _, tx := range pendingList {
|
||||||
c.markTxAsSeen(sender, tx)
|
c.markTxAsSeen(sender, tx)
|
||||||
}
|
}
|
||||||
c.sentTxLock.Unlock()
|
c.trackedTxLock.Unlock()
|
||||||
return proof.Nonce, pendingList, nil
|
return proof.Nonce, pendingList, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// allSenders returns the list of all senders that have belonging tracked transactions.
|
||||||
func (c *Client) allSenders() []common.Address {
|
func (c *Client) allSenders() []common.Address {
|
||||||
c.sentTxLock.Lock()
|
c.trackedTxLock.Lock()
|
||||||
allSenders := make([]common.Address, 0, len(c.sentTxs))
|
allSenders := make([]common.Address, 0, len(c.trackedTxs))
|
||||||
for sender := range c.sentTxs {
|
for sender := range c.trackedTxs {
|
||||||
allSenders = append(allSenders, sender)
|
allSenders = append(allSenders, sender)
|
||||||
}
|
}
|
||||||
c.sentTxLock.Unlock()
|
c.trackedTxLock.Unlock()
|
||||||
return allSenders
|
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) {
|
func (c *Client) txAndReceiptsNewHead(number uint64, hash common.Hash) {
|
||||||
c.sentTxLock.Lock()
|
c.trackedTxLock.Lock()
|
||||||
if number > c.lastHeadNumber {
|
if number > c.lastHeadNumber {
|
||||||
c.lastHeadNumber = number
|
c.lastHeadNumber = number
|
||||||
c.headCounter++
|
c.headCounter++
|
||||||
c.discardOldTxs()
|
c.discardOldTxs()
|
||||||
}
|
}
|
||||||
c.sentTxLock.Unlock()
|
c.trackedTxLock.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// discardOldTxs discards transactions that have not been seen as pending lately.
|
||||||
func (c *Client) discardOldTxs() {
|
func (c *Client) discardOldTxs() {
|
||||||
for sender, senderTxs := range c.sentTxs {
|
for sender, senderTxs := range c.trackedTxs {
|
||||||
for txHash, sentTx := range senderTxs {
|
for txHash, trackedTx := range senderTxs {
|
||||||
if sentTx.lastSeen+maxTxAge < c.headCounter {
|
if trackedTx.lastSeen+maxTxAge < c.headCounter {
|
||||||
delete(senderTxs, txHash)
|
delete(senderTxs, txHash)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if len(senderTxs) == 0 {
|
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) {
|
func (c *Client) markTxAsSeen(sender common.Address, tx *types.Transaction) {
|
||||||
senderTxs := c.sentTxs[sender]
|
senderTxs := c.trackedTxs[sender]
|
||||||
if senderTxs == nil {
|
if senderTxs == nil {
|
||||||
senderTxs = make(map[common.Hash]sentTx)
|
senderTxs = make(map[common.Hash]trackedTx)
|
||||||
c.sentTxs[sender] = senderTxs
|
c.trackedTxs[sender] = senderTxs
|
||||||
}
|
}
|
||||||
senderTxs[tx.Hash()] = sentTx{nonce: tx.Nonce(), lastSeen: c.headCounter}
|
senderTxs[tx.Hash()] = trackedTx{nonce: tx.Nonce(), lastSeen: c.headCounter}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue