mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 10:22:23 +00:00
ethclient/lightclient: refactored all functions into Client
This commit is contained in:
Zsolt Felfoldi
parent
5b05bb625d
commit
88a3179a8d
4 changed files with 251 additions and 298 deletions
|
|
@ -25,7 +25,6 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum"
|
||||
"github.com/ethereum/go-ethereum/beacon/light"
|
||||
"github.com/ethereum/go-ethereum/beacon/light/request"
|
||||
btypes "github.com/ethereum/go-ethereum/beacon/types"
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
|
|
@ -37,12 +36,8 @@ import (
|
|||
|
||||
const recentCanonicalLength = 256
|
||||
|
||||
type canonicalChain struct {
|
||||
lock sync.Mutex
|
||||
headTracker *light.HeadTracker
|
||||
blocksAndHeaders *blocksAndHeaders
|
||||
newHeadCb func(uint64, common.Hash)
|
||||
|
||||
type canonicalChainFields struct {
|
||||
chainLock sync.Mutex
|
||||
head, finality *btypes.ExecutionHeader
|
||||
recent map[uint64]common.Hash // nil until initialized
|
||||
recentTail uint64 // if recent != nil then recent hashes are available from recentTail to head
|
||||
|
|
@ -51,38 +46,32 @@ type canonicalChain struct {
|
|||
requests *requestMap[uint64, common.Hash] // requested; neither recent nor cached finalized
|
||||
}
|
||||
|
||||
func newCanonicalChain(headTracker *light.HeadTracker, blocksAndHeaders *blocksAndHeaders, newHeadCb func(uint64, common.Hash)) *canonicalChain {
|
||||
c := &canonicalChain{
|
||||
headTracker: headTracker,
|
||||
blocksAndHeaders: blocksAndHeaders,
|
||||
newHeadCb: newHeadCb,
|
||||
finalized: lru.NewCache[uint64, common.Hash](10000),
|
||||
requests: newRequestMap[uint64, common.Hash](nil),
|
||||
tailFetchCh: make(chan struct{}),
|
||||
}
|
||||
func (c *Client) initCanonicalChain() {
|
||||
c.finalized = lru.NewCache[uint64, common.Hash](10000)
|
||||
c.requests = newRequestMap[uint64, common.Hash](nil)
|
||||
c.tailFetchCh = make(chan struct{})
|
||||
go c.tailFetcher()
|
||||
return c
|
||||
}
|
||||
|
||||
// Process implements request.Module in order to get notified about new heads.
|
||||
func (c *canonicalChain) Process(requester request.Requester, events []request.Event) {
|
||||
func (c *Client) Process(requester request.Requester, events []request.Event) {
|
||||
if finality, ok := c.headTracker.ValidatedFinality(); ok {
|
||||
finalized := finality.Finalized.PayloadHeader
|
||||
c.setFinality(finalized)
|
||||
c.blocksAndHeaders.addPayloadHeader(finalized)
|
||||
c.addPayloadHeader(finalized)
|
||||
}
|
||||
if optimistic, ok := c.headTracker.ValidatedOptimistic(); ok {
|
||||
head := optimistic.Attested.PayloadHeader
|
||||
c.blocksAndHeaders.addPayloadHeader(head)
|
||||
c.addPayloadHeader(head)
|
||||
if c.setHead(head) {
|
||||
c.newHeadCb(head.BlockNumber(), head.BlockHash()) // should not block
|
||||
go c.processNewHead(head.BlockNumber(), head.BlockHash())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *canonicalChain) tailFetcher() { //TODO stop
|
||||
func (c *Client) tailFetcher() { //TODO stop
|
||||
for {
|
||||
c.lock.Lock()
|
||||
c.chainLock.Lock()
|
||||
var (
|
||||
tailNum uint64
|
||||
tailHash common.Hash
|
||||
|
|
@ -96,12 +85,12 @@ func (c *canonicalChain) tailFetcher() { //TODO stop
|
|||
needTail = reqNum
|
||||
}
|
||||
}
|
||||
c.lock.Unlock()
|
||||
c.chainLock.Unlock()
|
||||
if needTail < tailNum { //TODO check recentCanonicalLength
|
||||
log.Debug("Fetching tail headers", "have", tailNum, "need", needTail)
|
||||
ctx, _ := context.WithTimeout(context.Background(), time.Second*10)
|
||||
//TODO parallel fetch by number
|
||||
if header, err := c.blocksAndHeaders.getHeader(ctx, tailHash); err == nil {
|
||||
if header, err := c.getHeader(ctx, tailHash); err == nil {
|
||||
c.addRecentTail(header)
|
||||
}
|
||||
} else {
|
||||
|
|
@ -110,7 +99,7 @@ func (c *canonicalChain) tailFetcher() { //TODO stop
|
|||
}
|
||||
}
|
||||
|
||||
func (c *canonicalChain) getHash(ctx context.Context, number uint64) (common.Hash, error) {
|
||||
func (c *Client) getHash(ctx context.Context, number uint64) (common.Hash, error) {
|
||||
if hash, ok := c.getCachedHash(number); ok {
|
||||
return hash, nil
|
||||
}
|
||||
|
|
@ -123,19 +112,19 @@ func (c *canonicalChain) getHash(ctx context.Context, number uint64) (common.Has
|
|||
return req.getResult(ctx)
|
||||
}
|
||||
|
||||
func (c *canonicalChain) getCachedHash(number uint64) (common.Hash, bool) {
|
||||
c.lock.Lock()
|
||||
func (c *Client) getCachedHash(number uint64) (common.Hash, bool) {
|
||||
c.chainLock.Lock()
|
||||
hash, ok := c.recent[number]
|
||||
c.lock.Unlock()
|
||||
c.chainLock.Unlock()
|
||||
if ok {
|
||||
return hash, true
|
||||
}
|
||||
return c.finalized.Get(number)
|
||||
}
|
||||
|
||||
func (c *canonicalChain) setHead(head *btypes.ExecutionHeader) bool {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
func (c *Client) setHead(head *btypes.ExecutionHeader) bool {
|
||||
c.chainLock.Lock()
|
||||
defer c.chainLock.Unlock()
|
||||
|
||||
headNum, headHash := head.BlockNumber(), head.BlockHash()
|
||||
if c.head != nil && c.head.BlockHash() == headHash {
|
||||
|
|
@ -166,9 +155,9 @@ func (c *canonicalChain) setHead(head *btypes.ExecutionHeader) bool {
|
|||
return true
|
||||
}
|
||||
|
||||
func (c *canonicalChain) setFinality(finality *btypes.ExecutionHeader) {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
func (c *Client) setFinality(finality *btypes.ExecutionHeader) {
|
||||
c.chainLock.Lock()
|
||||
defer c.chainLock.Unlock()
|
||||
|
||||
c.finality = finality
|
||||
finalNum := finality.BlockNumber()
|
||||
|
|
@ -178,9 +167,9 @@ func (c *canonicalChain) setFinality(finality *btypes.ExecutionHeader) {
|
|||
c.requests.tryDeliver(finalNum, finality.BlockHash())
|
||||
}
|
||||
|
||||
func (c *canonicalChain) addRecentTail(tail *types.Header) bool {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
func (c *Client) addRecentTail(tail *types.Header) bool {
|
||||
c.chainLock.Lock()
|
||||
defer c.chainLock.Unlock()
|
||||
|
||||
if c.recent == nil || tail.Number.Uint64() != c.recentTail || c.recent[c.recentTail] != tail.Hash() {
|
||||
return false
|
||||
|
|
@ -193,21 +182,21 @@ func (c *canonicalChain) addRecentTail(tail *types.Header) bool {
|
|||
return true
|
||||
}
|
||||
|
||||
func (c *canonicalChain) getHead() *btypes.ExecutionHeader {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
func (c *Client) getHead() *btypes.ExecutionHeader {
|
||||
c.chainLock.Lock()
|
||||
defer c.chainLock.Unlock()
|
||||
|
||||
return c.head
|
||||
}
|
||||
|
||||
func (c *canonicalChain) getFinality() *btypes.ExecutionHeader {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
func (c *Client) getFinality() *btypes.ExecutionHeader {
|
||||
c.chainLock.Lock()
|
||||
defer c.chainLock.Unlock()
|
||||
|
||||
return c.finality
|
||||
}
|
||||
|
||||
func (c *canonicalChain) resolveBlockNumber(number *big.Int) (uint64, *btypes.ExecutionHeader, error) {
|
||||
func (c *Client) resolveBlockNumber(number *big.Int) (uint64, *btypes.ExecutionHeader, error) {
|
||||
if !number.IsInt64() {
|
||||
return 0, nil, errors.New("Invalid block number")
|
||||
}
|
||||
|
|
@ -231,7 +220,7 @@ func (c *canonicalChain) resolveBlockNumber(number *big.Int) (uint64, *btypes.Ex
|
|||
return uint64(num), nil, nil
|
||||
}
|
||||
|
||||
func (c *canonicalChain) blockNumberToHash(ctx context.Context, number *big.Int) (common.Hash, error) {
|
||||
func (c *Client) blockNumberToHash(ctx context.Context, number *big.Int) (common.Hash, error) {
|
||||
num, pheader, err := c.resolveBlockNumber(number)
|
||||
if err != nil {
|
||||
return common.Hash{}, err
|
||||
|
|
@ -242,13 +231,13 @@ func (c *canonicalChain) blockNumberToHash(ctx context.Context, number *big.Int)
|
|||
return c.getHash(ctx, num)
|
||||
}
|
||||
|
||||
func (c *canonicalChain) blockNumberOrHashToHash(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) (common.Hash, error) {
|
||||
func (c *Client) blockNumberOrHashToHash(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) (common.Hash, error) {
|
||||
if blockNrOrHash.BlockNumber != nil {
|
||||
return c.blockNumberToHash(ctx, big.NewInt(int64(*blockNrOrHash.BlockNumber)))
|
||||
}
|
||||
hash := *blockNrOrHash.BlockHash
|
||||
if blockNrOrHash.RequireCanonical {
|
||||
header, err := c.blocksAndHeaders.getHeader(ctx, hash)
|
||||
header, err := c.getHeader(ctx, hash)
|
||||
if err != nil {
|
||||
return common.Hash{}, err
|
||||
}
|
||||
|
|
@ -263,9 +252,7 @@ func (c *canonicalChain) blockNumberOrHashToHash(ctx context.Context, blockNrOrH
|
|||
return hash, nil
|
||||
}
|
||||
|
||||
type blocksAndHeaders struct {
|
||||
client *rpc.Client
|
||||
txAndReceipts *txAndReceipts
|
||||
type blocksAndHeadersFields struct {
|
||||
headerCache *lru.Cache[common.Hash, *types.Header]
|
||||
headerRequests *requestMap[common.Hash, *types.Header]
|
||||
payloadHeaderCache *lru.Cache[common.Hash, *btypes.ExecutionHeader]
|
||||
|
|
@ -273,22 +260,18 @@ type blocksAndHeaders struct {
|
|||
blockRequests *requestMap[common.Hash, *types.Block]
|
||||
}
|
||||
|
||||
func newBlocksAndHeaders(client *rpc.Client) *blocksAndHeaders {
|
||||
b := &blocksAndHeaders{
|
||||
client: client,
|
||||
headerCache: lru.NewCache[common.Hash, *types.Header](1000),
|
||||
payloadHeaderCache: lru.NewCache[common.Hash, *btypes.ExecutionHeader](1000),
|
||||
blockCache: lru.NewCache[common.Hash, *types.Block](10),
|
||||
}
|
||||
b.headerRequests = newRequestMap[common.Hash, *types.Header](b.requestHeader)
|
||||
b.blockRequests = newRequestMap[common.Hash, *types.Block](b.requestBlock)
|
||||
return b
|
||||
func (c *Client) initBlocksAndHeaders() {
|
||||
c.headerCache = lru.NewCache[common.Hash, *types.Header](1000)
|
||||
c.payloadHeaderCache = lru.NewCache[common.Hash, *btypes.ExecutionHeader](1000)
|
||||
c.blockCache = lru.NewCache[common.Hash, *types.Block](10)
|
||||
c.headerRequests = newRequestMap[common.Hash, *types.Header](c.requestHeader)
|
||||
c.blockRequests = newRequestMap[common.Hash, *types.Block](c.requestBlock)
|
||||
}
|
||||
|
||||
func (b *blocksAndHeaders) 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
|
||||
log.Debug("Starting RPC request", "type", "eth_getBlockByHash", "hash", hash, "full", false)
|
||||
err := b.client.CallContext(ctx, &header, "eth_getBlockByHash", hash, false)
|
||||
err := c.client.CallContext(ctx, &header, "eth_getBlockByHash", hash, false)
|
||||
if err == nil && header.Hash() != hash {
|
||||
header, err = nil, errors.New("header hash does not match")
|
||||
}
|
||||
|
|
@ -296,13 +279,13 @@ func (b *blocksAndHeaders) requestHeader(ctx context.Context, hash common.Hash)
|
|||
return header, err
|
||||
}
|
||||
|
||||
func (b *blocksAndHeaders) requestBlock(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
func (c *Client) requestBlock(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
var (
|
||||
raw json.RawMessage
|
||||
block *types.Block
|
||||
)
|
||||
log.Debug("Starting RPC request", "type", "eth_getBlockByHash", "hash", hash, "full", true)
|
||||
err := b.client.CallContext(ctx, &raw, "eth_getBlockByHash", hash, true)
|
||||
err := c.client.CallContext(ctx, &raw, "eth_getBlockByHash", hash, true)
|
||||
log.Debug("Finished RPC request", "type", "eth_getBlockByHash", "hash", hash, "full", true, "error", err)
|
||||
if err == nil {
|
||||
block, err = decodeBlock(raw)
|
||||
|
|
@ -313,20 +296,20 @@ func (b *blocksAndHeaders) requestBlock(ctx context.Context, hash common.Hash) (
|
|||
return block, err
|
||||
}
|
||||
|
||||
func (b *blocksAndHeaders) getHeader(ctx context.Context, hash common.Hash) (*types.Header, error) {
|
||||
if header, ok := b.headerCache.Get(hash); ok {
|
||||
func (c *Client) getHeader(ctx context.Context, hash common.Hash) (*types.Header, error) {
|
||||
if header, ok := c.headerCache.Get(hash); ok {
|
||||
return header, nil
|
||||
}
|
||||
if block, ok := b.blockCache.Get(hash); ok {
|
||||
if block, ok := c.blockCache.Get(hash); ok {
|
||||
return block.Header(), nil
|
||||
}
|
||||
if b.blockRequests.has(hash) && !b.headerRequests.has(hash) {
|
||||
req := b.blockRequests.request(hash)
|
||||
if c.blockRequests.has(hash) && !c.headerRequests.has(hash) {
|
||||
req := c.blockRequests.request(hash)
|
||||
block, err := req.getResult(ctx)
|
||||
if err == nil {
|
||||
header := block.Header()
|
||||
b.headerCache.Add(hash, header)
|
||||
b.blockCache.Add(hash, block)
|
||||
c.headerCache.Add(hash, header)
|
||||
c.blockCache.Add(hash, block)
|
||||
req.release()
|
||||
return header, nil
|
||||
} else {
|
||||
|
|
@ -334,32 +317,32 @@ func (b *blocksAndHeaders) getHeader(ctx context.Context, hash common.Hash) (*ty
|
|||
return nil, err
|
||||
}
|
||||
}
|
||||
req := b.headerRequests.request(hash)
|
||||
req := c.headerRequests.request(hash)
|
||||
header, err := req.getResult(ctx)
|
||||
if err == nil {
|
||||
b.headerCache.Add(hash, header)
|
||||
c.headerCache.Add(hash, header)
|
||||
}
|
||||
req.release()
|
||||
return header, err
|
||||
}
|
||||
|
||||
func (b *blocksAndHeaders) getPayloadHeader(hash common.Hash) *btypes.ExecutionHeader {
|
||||
pheader, _ := b.payloadHeaderCache.Get(hash)
|
||||
func (c *Client) getPayloadHeader(hash common.Hash) *btypes.ExecutionHeader {
|
||||
pheader, _ := c.payloadHeaderCache.Get(hash)
|
||||
return pheader
|
||||
}
|
||||
|
||||
func (b *blocksAndHeaders) getBlock(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
if block, ok := b.blockCache.Get(hash); ok {
|
||||
func (c *Client) getBlock(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
if block, ok := c.blockCache.Get(hash); ok {
|
||||
return block, nil
|
||||
}
|
||||
req := b.blockRequests.request(hash)
|
||||
req := c.blockRequests.request(hash)
|
||||
block, err := req.getResult(ctx)
|
||||
if err == nil {
|
||||
header := block.Header()
|
||||
b.headerCache.Add(hash, header)
|
||||
b.headerRequests.tryDeliver(hash, header)
|
||||
b.blockCache.Add(hash, block)
|
||||
b.txAndReceipts.cacheBlockTxPositions(block)
|
||||
c.headerCache.Add(hash, header)
|
||||
c.headerRequests.tryDeliver(hash, header)
|
||||
c.blockCache.Add(hash, block)
|
||||
c.cacheBlockTxPositions(block)
|
||||
}
|
||||
req.release()
|
||||
return block, err
|
||||
|
|
@ -467,6 +450,6 @@ func decodeBlock(raw json.RawMessage) (*types.Block, error) {
|
|||
return types.NewBlockWithHeader(head).WithBody(types.Body{Transactions: txs, Withdrawals: body.Withdrawals}), nil
|
||||
}
|
||||
|
||||
func (b *blocksAndHeaders) addPayloadHeader(header *btypes.ExecutionHeader) {
|
||||
b.payloadHeaderCache.Add(header.BlockHash(), header)
|
||||
func (c *Client) addPayloadHeader(header *btypes.ExecutionHeader) {
|
||||
c.payloadHeaderCache.Add(header.BlockHash(), header)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,13 +38,18 @@ import (
|
|||
)
|
||||
|
||||
type Client struct {
|
||||
clConfig config.LightClientConfig
|
||||
elConfig *params.ChainConfig
|
||||
scheduler *request.Scheduler
|
||||
canonicalChain *canonicalChain
|
||||
blocksAndHeaders *blocksAndHeaders
|
||||
txAndReceipts *txAndReceipts
|
||||
state *lightState
|
||||
canonicalChainFields
|
||||
blocksAndHeadersFields
|
||||
lightStateFields
|
||||
txAndReceiptsFields
|
||||
|
||||
clConfig config.LightClientConfig
|
||||
elConfig *params.ChainConfig
|
||||
|
||||
client *rpc.Client
|
||||
scheduler *request.Scheduler
|
||||
headTracker *light.HeadTracker
|
||||
|
||||
headSubLock ssync.Mutex
|
||||
headSubs map[*headSub]struct{}
|
||||
cancelHeadFetch func()
|
||||
|
|
@ -53,38 +58,29 @@ type Client struct {
|
|||
|
||||
func NewClient(clConfig config.LightClientConfig, elConfig *params.ChainConfig, db ethdb.KeyValueStore, rpcClient *rpc.Client) *Client {
|
||||
// create data structures
|
||||
var (
|
||||
committeeChain = light.NewCommitteeChain(db, clConfig.ChainConfig, clConfig.SignerThreshold, clConfig.EnforceTime)
|
||||
headTracker = light.NewHeadTracker(committeeChain, clConfig.SignerThreshold)
|
||||
)
|
||||
// set up scheduler and sync modules
|
||||
//chainHeadFeed := new(event.Feed)
|
||||
scheduler := request.NewScheduler()
|
||||
committeeChain := light.NewCommitteeChain(db, clConfig.ChainConfig, clConfig.SignerThreshold, clConfig.EnforceTime)
|
||||
client := &Client{
|
||||
clConfig: clConfig,
|
||||
elConfig: elConfig,
|
||||
scheduler: scheduler,
|
||||
headSubs: make(map[*headSub]struct{}),
|
||||
client: rpcClient,
|
||||
clConfig: clConfig,
|
||||
elConfig: elConfig,
|
||||
scheduler: request.NewScheduler(),
|
||||
headTracker: light.NewHeadTracker(committeeChain, clConfig.SignerThreshold),
|
||||
headSubs: make(map[*headSub]struct{}),
|
||||
}
|
||||
blocksAndHeaders := newBlocksAndHeaders(rpcClient)
|
||||
canonicalChain := newCanonicalChain(headTracker, blocksAndHeaders, client.newHead)
|
||||
txAndReceipts := newTxAndReceipts(rpcClient, canonicalChain, blocksAndHeaders, elConfig)
|
||||
blocksAndHeaders.txAndReceipts = txAndReceipts
|
||||
client.blocksAndHeaders = blocksAndHeaders
|
||||
client.txAndReceipts = txAndReceipts
|
||||
client.canonicalChain = canonicalChain
|
||||
client.state = newLightState(rpcClient, canonicalChain, blocksAndHeaders)
|
||||
txAndReceipts.lightState = client.state //TODO init these refs in a nicer way???
|
||||
client.initBlocksAndHeaders()
|
||||
client.initCanonicalChain()
|
||||
client.initTxAndReceipts()
|
||||
client.initLightState()
|
||||
|
||||
checkpointInit := sync.NewCheckpointInit(committeeChain, clConfig.Checkpoint)
|
||||
forwardSync := sync.NewForwardUpdateSync(committeeChain)
|
||||
headSync := sync.NewHeadSync(headTracker, committeeChain)
|
||||
scheduler.RegisterTarget(headTracker)
|
||||
scheduler.RegisterTarget(committeeChain)
|
||||
scheduler.RegisterModule(checkpointInit, "checkpointInit")
|
||||
scheduler.RegisterModule(forwardSync, "forwardSync")
|
||||
scheduler.RegisterModule(headSync, "headSync")
|
||||
scheduler.RegisterModule(client.canonicalChain, "canonicalChain")
|
||||
headSync := sync.NewHeadSync(client.headTracker, committeeChain)
|
||||
client.scheduler.RegisterTarget(client.headTracker)
|
||||
client.scheduler.RegisterTarget(committeeChain)
|
||||
client.scheduler.RegisterModule(checkpointInit, "checkpointInit")
|
||||
client.scheduler.RegisterModule(forwardSync, "forwardSync")
|
||||
client.scheduler.RegisterModule(headSync, "headSync")
|
||||
client.scheduler.RegisterModule(client, "canonicalChain")
|
||||
return client
|
||||
}
|
||||
|
||||
|
|
@ -105,11 +101,11 @@ func (c *Client) Stop() {
|
|||
// ChainReader interface
|
||||
|
||||
func (c *Client) BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
return c.blocksAndHeaders.getBlock(ctx, hash)
|
||||
return c.getBlock(ctx, hash)
|
||||
}
|
||||
|
||||
func (c *Client) BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error) {
|
||||
hash, err := c.canonicalChain.blockNumberToHash(ctx, number)
|
||||
hash, err := c.blockNumberToHash(ctx, number)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -117,11 +113,11 @@ func (c *Client) BlockByNumber(ctx context.Context, number *big.Int) (*types.Blo
|
|||
}
|
||||
|
||||
func (c *Client) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
|
||||
return c.blocksAndHeaders.getHeader(ctx, hash)
|
||||
return c.getHeader(ctx, hash)
|
||||
}
|
||||
|
||||
func (c *Client) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
|
||||
hash, err := c.canonicalChain.blockNumberToHash(ctx, number)
|
||||
hash, err := c.blockNumberToHash(ctx, number)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -160,36 +156,34 @@ func (c *Client) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header)
|
|||
return sub, nil
|
||||
}
|
||||
|
||||
func (c *Client) newHead(number uint64, hash common.Hash) {
|
||||
go func() {
|
||||
log.Trace("New execution payload header received", "hash", hash)
|
||||
c.txAndReceipts.newHead(number, hash)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
c.headSubLock.Lock()
|
||||
if len(c.headSubs) == 0 {
|
||||
c.headSubLock.Unlock()
|
||||
return
|
||||
}
|
||||
if c.cancelHeadFetch != nil {
|
||||
c.cancelHeadFetch()
|
||||
}
|
||||
c.cancelHeadFetch = cancel
|
||||
c.headFetchCounter++
|
||||
hfc := c.headFetchCounter
|
||||
func (c *Client) processNewHead(number uint64, hash common.Hash) {
|
||||
log.Trace("New execution payload header received", "hash", hash)
|
||||
c.txAndReceiptsNewHead(number, hash)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
c.headSubLock.Lock()
|
||||
if len(c.headSubs) == 0 {
|
||||
c.headSubLock.Unlock()
|
||||
return
|
||||
}
|
||||
if c.cancelHeadFetch != nil {
|
||||
c.cancelHeadFetch()
|
||||
}
|
||||
c.cancelHeadFetch = cancel
|
||||
c.headFetchCounter++
|
||||
hfc := c.headFetchCounter
|
||||
c.headSubLock.Unlock()
|
||||
|
||||
head, err := c.blocksAndHeaders.getHeader(ctx, hash)
|
||||
c.headSubLock.Lock()
|
||||
if c.headFetchCounter == hfc {
|
||||
c.cancelHeadFetch = nil
|
||||
head, err := c.getHeader(ctx, hash)
|
||||
c.headSubLock.Lock()
|
||||
if c.headFetchCounter == hfc {
|
||||
c.cancelHeadFetch = nil
|
||||
}
|
||||
if err == nil {
|
||||
for sub := range c.headSubs {
|
||||
sub.headCh <- head
|
||||
}
|
||||
if err == nil {
|
||||
for sub := range c.headSubs {
|
||||
sub.headCh <- head
|
||||
}
|
||||
}
|
||||
c.headSubLock.Unlock()
|
||||
}()
|
||||
}
|
||||
c.headSubLock.Unlock()
|
||||
}
|
||||
|
||||
func (c *Client) unsubscribeNewHead(sub *headSub) {
|
||||
|
|
@ -216,17 +210,17 @@ func (h *headSub) Err() <-chan error {
|
|||
// TransactionReader interface
|
||||
|
||||
func (c *Client) TransactionByHash(ctx context.Context, txHash common.Hash) (tx *types.Transaction, isPending bool, err error) {
|
||||
return c.txAndReceipts.getTxByHash(ctx, txHash)
|
||||
return c.getTxByHash(ctx, txHash)
|
||||
}
|
||||
|
||||
func (c *Client) TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
|
||||
return c.txAndReceipts.getReceiptByTxHash(ctx, txHash)
|
||||
return c.getReceiptByTxHash(ctx, txHash)
|
||||
}
|
||||
|
||||
// ChainStateReader interface
|
||||
|
||||
func (c *Client) BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error) {
|
||||
proof, _, err := c.state.getProof(ctx, blockNumber, account, nil, false)
|
||||
proof, _, err := c.getProof(ctx, blockNumber, account, nil, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -234,7 +228,7 @@ func (c *Client) BalanceAt(ctx context.Context, account common.Address, blockNum
|
|||
}
|
||||
|
||||
func (c *Client) StorageAt(ctx context.Context, account common.Address, key common.Hash, blockNumber *big.Int) ([]byte, error) {
|
||||
proof, _, err := c.state.getProof(ctx, blockNumber, account, []string{key.Hex()}, false)
|
||||
proof, _, err := c.getProof(ctx, blockNumber, account, []string{key.Hex()}, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -242,7 +236,7 @@ func (c *Client) StorageAt(ctx context.Context, account common.Address, key comm
|
|||
}
|
||||
|
||||
func (c *Client) CodeAt(ctx context.Context, account common.Address, blockNumber *big.Int) ([]byte, error) {
|
||||
_, code, err := c.state.getProof(ctx, blockNumber, account, nil, true)
|
||||
_, code, err := c.getProof(ctx, blockNumber, account, nil, true)
|
||||
return code, err
|
||||
}
|
||||
|
||||
|
|
@ -250,7 +244,7 @@ func (c *Client) NonceAt(ctx context.Context, account common.Address, blockNumbe
|
|||
if blockNumber.IsInt64() && rpc.BlockNumber(blockNumber.Int64()) == rpc.PendingBlockNumber {
|
||||
return c.PendingNonceAt(ctx, account)
|
||||
}
|
||||
proof, _, err := c.state.getProof(ctx, blockNumber, account, nil, false)
|
||||
proof, _, err := c.getProof(ctx, blockNumber, account, nil, false)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
|
@ -258,17 +252,17 @@ func (c *Client) NonceAt(ctx context.Context, account common.Address, blockNumbe
|
|||
}
|
||||
|
||||
func (c *Client) BlockReceipts(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) ([]*types.Receipt, error) {
|
||||
hash, err := c.canonicalChain.blockNumberOrHashToHash(ctx, blockNrOrHash)
|
||||
hash, err := c.blockNumberOrHashToHash(ctx, blockNrOrHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.txAndReceipts.getBlockReceipts(ctx, hash)
|
||||
return c.getBlockReceipts(ctx, hash)
|
||||
}
|
||||
|
||||
// TransactionSender interface
|
||||
|
||||
func (c *Client) SendTransaction(ctx context.Context, tx *types.Transaction) error {
|
||||
return c.txAndReceipts.sendTransaction(ctx, tx)
|
||||
return c.sendTransaction(ctx, tx)
|
||||
}
|
||||
|
||||
// PendingStateReader interface
|
||||
|
|
@ -287,11 +281,11 @@ func (c *Client) PendingCodeAt(ctx context.Context, account common.Address) ([]b
|
|||
}
|
||||
|
||||
func (c *Client) PendingNonceAt(ctx context.Context, account common.Address) (uint64, error) {
|
||||
head := c.canonicalChain.getHead()
|
||||
head := c.getHead()
|
||||
if head == nil {
|
||||
return 0, errors.New("chain head unknown")
|
||||
}
|
||||
headNonce, pendingTxs, err := c.txAndReceipts.nonceAndPendingTxs(ctx, head, account)
|
||||
headNonce, pendingTxs, err := c.nonceAndPendingTxs(ctx, head, account)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
|
@ -302,16 +296,16 @@ func (c *Client) PendingNonceAt(ctx context.Context, account common.Address) (ui
|
|||
}
|
||||
|
||||
func (c *Client) PendingTransactionCount(ctx context.Context) (uint, error) {
|
||||
head := c.canonicalChain.getHead()
|
||||
head := c.getHead()
|
||||
if head == nil {
|
||||
return 0, errors.New("chain head unknown")
|
||||
}
|
||||
allSenders := c.txAndReceipts.allSenders()
|
||||
allSenders := c.allSenders()
|
||||
countCh := make(chan uint, len(allSenders))
|
||||
errCh := make(chan error, len(allSenders))
|
||||
for _, sender := range allSenders {
|
||||
go func() {
|
||||
_, pendingTxs, err := c.txAndReceipts.nonceAndPendingTxs(ctx, head, sender)
|
||||
_, pendingTxs, err := c.nonceAndPendingTxs(ctx, head, sender)
|
||||
errCh <- err
|
||||
countCh <- uint(len(pendingTxs))
|
||||
}()
|
||||
|
|
@ -329,7 +323,7 @@ func (c *Client) PendingTransactionCount(ctx context.Context) (uint, error) {
|
|||
// BlockNumberReader interface
|
||||
|
||||
func (c *Client) BlockNumber(ctx context.Context) (uint64, error) {
|
||||
if head := c.canonicalChain.getHead(); head != nil {
|
||||
if head := c.getHead(); head != nil {
|
||||
return head.BlockNumber(), nil
|
||||
}
|
||||
return 0, errors.New("chain head unknown")
|
||||
|
|
|
|||
|
|
@ -31,7 +31,6 @@ import (
|
|||
"github.com/ethereum/go-ethereum/ethclient/gethclient"
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/rlp"
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
"github.com/ethereum/go-ethereum/trie"
|
||||
"github.com/holiman/uint256"
|
||||
)
|
||||
|
|
@ -47,43 +46,34 @@ type codeRequest struct {
|
|||
address common.Address
|
||||
}
|
||||
|
||||
type lightState struct {
|
||||
client *rpc.Client
|
||||
canonicalChain *canonicalChain
|
||||
blocksAndHeaders *blocksAndHeaders
|
||||
proofCache *lru.Cache[proofRequest, *gethclient.AccountResult]
|
||||
proofRequests *requestMap[proofRequest, *gethclient.AccountResult]
|
||||
codeCache *lru.Cache[codeRequest, []byte]
|
||||
codeRequests *requestMap[codeRequest, []byte]
|
||||
type lightStateFields struct {
|
||||
proofCache *lru.Cache[proofRequest, *gethclient.AccountResult]
|
||||
proofRequests *requestMap[proofRequest, *gethclient.AccountResult]
|
||||
codeCache *lru.Cache[codeRequest, []byte]
|
||||
codeRequests *requestMap[codeRequest, []byte]
|
||||
}
|
||||
|
||||
func newLightState(client *rpc.Client, canonicalChain *canonicalChain, blocksAndHeaders *blocksAndHeaders) *lightState {
|
||||
s := &lightState{
|
||||
client: client,
|
||||
canonicalChain: canonicalChain,
|
||||
blocksAndHeaders: blocksAndHeaders,
|
||||
proofCache: lru.NewCache[proofRequest, *gethclient.AccountResult](100),
|
||||
codeCache: lru.NewCache[codeRequest, []byte](10),
|
||||
}
|
||||
s.proofRequests = newRequestMap[proofRequest, *gethclient.AccountResult](s.requestProof)
|
||||
s.codeRequests = newRequestMap[codeRequest, []byte](s.requestCode)
|
||||
return s
|
||||
func (c *Client) initLightState() {
|
||||
c.proofCache = lru.NewCache[proofRequest, *gethclient.AccountResult](100)
|
||||
c.codeCache = lru.NewCache[codeRequest, []byte](10)
|
||||
c.proofRequests = newRequestMap[proofRequest, *gethclient.AccountResult](c.requestProof)
|
||||
c.codeRequests = newRequestMap[codeRequest, []byte](c.requestCode)
|
||||
}
|
||||
|
||||
func (s *lightState) fetchProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) {
|
||||
if proof, ok := s.proofCache.Get(req); ok {
|
||||
func (c *Client) fetchProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) {
|
||||
if proof, ok := c.proofCache.Get(req); ok {
|
||||
return proof, nil
|
||||
}
|
||||
request := s.proofRequests.request(req)
|
||||
request := c.proofRequests.request(req)
|
||||
proof, err := request.getResult(ctx)
|
||||
if err == nil {
|
||||
s.proofCache.Add(req, proof) //TODO cached before validation; remove and retry if invalid
|
||||
c.proofCache.Add(req, proof) //TODO cached before validation; remove and retry if invalid
|
||||
}
|
||||
request.release()
|
||||
return proof, err
|
||||
}
|
||||
|
||||
func (s *lightState) requestProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) {
|
||||
func (c *Client) requestProof(ctx context.Context, req proofRequest) (*gethclient.AccountResult, error) {
|
||||
type storageResult struct {
|
||||
Key string `json:"key"`
|
||||
Value *hexutil.Big `json:"value"`
|
||||
|
|
@ -106,7 +96,7 @@ func (s *lightState) requestProof(ctx context.Context, req proofRequest) (*gethc
|
|||
}
|
||||
log.Debug("Starting RPC request", "type", "eth_getProof", "blockNumber", req.blockNumber, "address", req.address, "storageKeys", len(storageKeys))
|
||||
var res accountResult
|
||||
err := s.client.CallContext(ctx, &res, "eth_getProof", req.address, storageKeys, hexutil.EncodeUint64(req.blockNumber))
|
||||
err := c.client.CallContext(ctx, &res, "eth_getProof", req.address, storageKeys, hexutil.EncodeUint64(req.blockNumber))
|
||||
log.Debug("Finished RPC request", "type", "eth_getProof", "blockNumber", req.blockNumber, "address", req.address, "storageKeys", len(storageKeys), "error", err)
|
||||
var proof *gethclient.AccountResult
|
||||
if err == nil { //TODO de-duplicate
|
||||
|
|
@ -132,23 +122,23 @@ func (s *lightState) requestProof(ctx context.Context, req proofRequest) (*gethc
|
|||
return proof, err
|
||||
}
|
||||
|
||||
func (s *lightState) fetchCode(ctx context.Context, req codeRequest) ([]byte, error) {
|
||||
if code, ok := s.codeCache.Get(req); ok {
|
||||
func (c *Client) fetchCode(ctx context.Context, req codeRequest) ([]byte, error) {
|
||||
if code, ok := c.codeCache.Get(req); ok {
|
||||
return code, nil
|
||||
}
|
||||
request := s.codeRequests.request(req)
|
||||
request := c.codeRequests.request(req)
|
||||
code, err := request.getResult(ctx)
|
||||
if err == nil {
|
||||
s.codeCache.Add(req, code) //TODO cached before validation; remove and retry if invalid
|
||||
c.codeCache.Add(req, code) //TODO cached before validation; remove and retry if invalid
|
||||
}
|
||||
request.release()
|
||||
return code, err
|
||||
}
|
||||
|
||||
func (s *lightState) requestCode(ctx context.Context, req codeRequest) ([]byte, error) {
|
||||
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)
|
||||
err := s.client.CallContext(ctx, &code, "eth_getCode", req.address, hexutil.EncodeUint64(req.blockNumber))
|
||||
err := c.client.CallContext(ctx, &code, "eth_getCode", req.address, hexutil.EncodeUint64(req.blockNumber))
|
||||
log.Debug("Finished RPC request", "type", "eth_getCode", "blockNumber", req.blockNumber, "address", req.address, "error", err)
|
||||
return code, err
|
||||
}
|
||||
|
|
@ -199,8 +189,8 @@ func stValueBytes(value *big.Int) ([]byte, error) {
|
|||
}
|
||||
}
|
||||
|
||||
func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account common.Address, storageKeys []string, getCode bool) (*gethclient.AccountResult, []byte, error) {
|
||||
num, pheader, err := s.canonicalChain.resolveBlockNumber(blockNumber)
|
||||
func (c *Client) getProof(ctx context.Context, blockNumber *big.Int, account common.Address, storageKeys []string, getCode bool) (*gethclient.AccountResult, []byte, error) {
|
||||
num, pheader, err := c.resolveBlockNumber(blockNumber)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
|
@ -216,16 +206,16 @@ func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account
|
|||
go func() {
|
||||
defer close(stateRootCh)
|
||||
|
||||
blockHash, err := s.canonicalChain.getHash(ctx, num)
|
||||
blockHash, err := c.getHash(ctx, num)
|
||||
if err != nil {
|
||||
stateRootErr = err
|
||||
return
|
||||
}
|
||||
if pheader := s.blocksAndHeaders.getPayloadHeader(blockHash); pheader != nil {
|
||||
if pheader := c.getPayloadHeader(blockHash); pheader != nil {
|
||||
stateRoot = pheader.StateRoot()
|
||||
return
|
||||
}
|
||||
header, err := s.blocksAndHeaders.getHeader(ctx, blockHash)
|
||||
header, err := c.getHeader(ctx, blockHash)
|
||||
if err != nil {
|
||||
stateRootErr = err
|
||||
return
|
||||
|
|
@ -240,11 +230,11 @@ func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account
|
|||
)
|
||||
if getCode {
|
||||
go func() {
|
||||
code, codeErr = s.fetchCode(ctx, codeRequest{blockNumber: num, address: account})
|
||||
code, codeErr = c.fetchCode(ctx, codeRequest{blockNumber: num, address: account})
|
||||
close(codeCh)
|
||||
}()
|
||||
}
|
||||
proof, proofErr := s.fetchProof(ctx, proofRequest{blockNumber: num, address: account, storageKeys: strings.Join(storageKeys, ",")})
|
||||
proof, proofErr := c.fetchProof(ctx, proofRequest{blockNumber: num, address: account, storageKeys: strings.Join(storageKeys, ",")})
|
||||
if proofErr != nil {
|
||||
return nil, nil, proofErr
|
||||
}
|
||||
|
|
@ -252,7 +242,7 @@ func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account
|
|||
if stateRootErr != nil {
|
||||
return nil, nil, stateRootErr
|
||||
}
|
||||
if err := s.validateProof(proof, stateRoot, account, storageKeys); err != nil {
|
||||
if err := c.validateProof(proof, stateRoot, account, storageKeys); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if getCode {
|
||||
|
|
@ -267,7 +257,7 @@ func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account
|
|||
return proof, code, nil
|
||||
}
|
||||
|
||||
func (s *lightState) 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
|
||||
proofReader, err := makeProofReader(proof.AccountProof)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -32,20 +32,13 @@ import (
|
|||
"github.com/ethereum/go-ethereum/common/lru"
|
||||
"github.com/ethereum/go-ethereum/consensus/misc/eip4844"
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
"github.com/ethereum/go-ethereum/params"
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
"github.com/ethereum/go-ethereum/trie"
|
||||
)
|
||||
|
||||
const maxTxAge = 300 // sent transactions are typically remembered for an hour after last seen pending
|
||||
|
||||
type txAndReceipts struct {
|
||||
client *rpc.Client
|
||||
canonicalChain *canonicalChain
|
||||
blocksAndHeaders *blocksAndHeaders
|
||||
lightState *lightState
|
||||
elConfig *params.ChainConfig
|
||||
signer types.Signer
|
||||
type txAndReceiptsFields struct {
|
||||
signer types.Signer
|
||||
|
||||
receiptsCache *lru.Cache[common.Hash, types.Receipts]
|
||||
receiptsRequests *requestMap[common.Hash, types.Receipts]
|
||||
|
|
@ -69,40 +62,33 @@ type sentTx struct {
|
|||
|
||||
type senderTxs map[common.Hash]sentTx
|
||||
|
||||
func newTxAndReceipts(client *rpc.Client, canonicalChain *canonicalChain, blocksAndHeaders *blocksAndHeaders, elConfig *params.ChainConfig) *txAndReceipts {
|
||||
t := &txAndReceipts{
|
||||
client: client,
|
||||
canonicalChain: canonicalChain,
|
||||
blocksAndHeaders: blocksAndHeaders,
|
||||
elConfig: elConfig,
|
||||
receiptsCache: lru.NewCache[common.Hash, types.Receipts](10),
|
||||
txPosCache: lru.NewCache[common.Hash, txInBlock](10000),
|
||||
sentTxs: make(map[common.Address]senderTxs),
|
||||
signer: types.LatestSigner(elConfig),
|
||||
}
|
||||
t.receiptsRequests = newRequestMap[common.Hash, types.Receipts](t.requestBlockReceipts)
|
||||
return t
|
||||
func (c *Client) initTxAndReceipts() {
|
||||
c.receiptsCache = lru.NewCache[common.Hash, types.Receipts](10)
|
||||
c.txPosCache = lru.NewCache[common.Hash, txInBlock](10000)
|
||||
c.sentTxs = make(map[common.Address]senderTxs)
|
||||
c.signer = types.LatestSigner(c.elConfig)
|
||||
c.receiptsRequests = newRequestMap[common.Hash, types.Receipts](c.requestBlockReceipts)
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) getTxByHash(ctx context.Context, txHash common.Hash) (tx *types.Transaction, isPending bool, err error) {
|
||||
if pos, ok := t.txPosCache.Get(txHash); ok {
|
||||
if hash, ok := t.canonicalChain.getCachedHash(pos.blockNumber); ok && hash == pos.blockHash {
|
||||
if block, ok := t.blocksAndHeaders.blockCache.Get(pos.blockHash); ok {
|
||||
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 {
|
||||
if block, ok := c.blockCache.Get(pos.blockHash); ok {
|
||||
return block.Transactions()[pos.index], false, nil //TODO index range check
|
||||
}
|
||||
}
|
||||
}
|
||||
var headBlockNumber uint64
|
||||
if head := t.canonicalChain.getHead(); head != nil {
|
||||
if head := c.getHead(); head != nil {
|
||||
headBlockNumber = head.BlockNumber()
|
||||
}
|
||||
return t.getUncachedTxByHash(ctx, txHash, headBlockNumber)
|
||||
return c.getUncachedTxByHash(ctx, txHash, headBlockNumber)
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) getUncachedTxByHash(ctx context.Context, txHash common.Hash, headBlockNumber uint64) (tx *types.Transaction, isPending bool, err error) {
|
||||
tx, isPending, err = t.requestTxByHash(ctx, txHash)
|
||||
func (c *Client) getUncachedTxByHash(ctx context.Context, txHash common.Hash, headBlockNumber uint64) (tx *types.Transaction, isPending bool, err error) {
|
||||
tx, isPending, err = c.requestTxByHash(ctx, txHash)
|
||||
if err == nil && !isPending {
|
||||
receipt, err := t.requestReceiptByTxHash(ctx, txHash)
|
||||
receipt, err := c.requestReceiptByTxHash(ctx, txHash)
|
||||
if err == ethereum.NotFound {
|
||||
return tx, false, nil
|
||||
}
|
||||
|
|
@ -117,15 +103,15 @@ func (t *txAndReceipts) getUncachedTxByHash(ctx context.Context, txHash common.H
|
|||
return
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) getReceiptByTxHash(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
|
||||
if pos, ok := t.txPosCache.Get(txHash); ok {
|
||||
if hash, ok := t.canonicalChain.getCachedHash(pos.blockNumber); ok && hash == pos.blockHash {
|
||||
if receipts, ok := t.receiptsCache.Get(pos.blockHash); ok {
|
||||
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 {
|
||||
if receipts, ok := c.receiptsCache.Get(pos.blockHash); ok {
|
||||
return receipts[pos.index], nil //TODO index range check
|
||||
}
|
||||
}
|
||||
}
|
||||
receipt, err := t.requestReceiptByTxHash(ctx, txHash)
|
||||
receipt, err := c.requestReceiptByTxHash(ctx, txHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -138,11 +124,11 @@ func (t *txAndReceipts) getReceiptByTxHash(ctx context.Context, txHash common.Ha
|
|||
return nil, errors.New("receipt references non-canonical block")
|
||||
}
|
||||
blockNumber := receipt.BlockNumber.Uint64()
|
||||
if head := t.canonicalChain.getHead(); head == nil || head.BlockNumber() < blockNumber {
|
||||
if head := c.getHead(); head == nil || head.BlockNumber() < blockNumber {
|
||||
// consider it pending if it's reported to be included higher than the light chain head
|
||||
return nil, ethereum.NotFound
|
||||
}
|
||||
canonicalHash, err := t.canonicalChain.getHash(ctx, blockNumber)
|
||||
canonicalHash, err := c.getHash(ctx, blockNumber)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -150,7 +136,7 @@ func (t *txAndReceipts) getReceiptByTxHash(ctx context.Context, txHash common.Ha
|
|||
return nil, errors.New("receipt references non-canonical block")
|
||||
}
|
||||
// check if it is the actual canonical receipt at the given position
|
||||
receipts, err := t.getBlockReceipts(ctx, receipt.BlockHash)
|
||||
receipts, err := c.getBlockReceipts(ctx, receipt.BlockHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -169,31 +155,31 @@ func (t *txAndReceipts) getReceiptByTxHash(ctx context.Context, txHash common.Ha
|
|||
if !bytes.Equal(jsonReceived, jsonCanonical) {
|
||||
return nil, errors.New("received and derived receipts do not match")
|
||||
}
|
||||
t.txPosCache.Add(receipt.TxHash, txInBlock{blockNumber: blockNumber, blockHash: receipt.BlockHash, index: receipt.TransactionIndex})
|
||||
c.txPosCache.Add(receipt.TxHash, txInBlock{blockNumber: blockNumber, blockHash: receipt.BlockHash, index: receipt.TransactionIndex})
|
||||
return receipt, err
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) getBlockReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) {
|
||||
if receipts, ok := t.receiptsCache.Get(blockHash); ok {
|
||||
func (c *Client) getBlockReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) {
|
||||
if receipts, ok := c.receiptsCache.Get(blockHash); ok {
|
||||
return receipts, nil
|
||||
}
|
||||
request := t.receiptsRequests.request(blockHash)
|
||||
block, err := t.blocksAndHeaders.getBlock(ctx, blockHash)
|
||||
request := c.receiptsRequests.request(blockHash)
|
||||
block, err := c.getBlock(ctx, blockHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
receipts, err := request.getResult(ctx)
|
||||
if err == nil {
|
||||
receipts, err = t.validateBlockReceipts(block, receipts)
|
||||
receipts, err = c.validateBlockReceipts(block, receipts)
|
||||
}
|
||||
if err == nil {
|
||||
t.receiptsCache.Add(blockHash, receipts)
|
||||
c.receiptsCache.Add(blockHash, receipts)
|
||||
}
|
||||
request.release()
|
||||
return receipts, err
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) 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
|
||||
var hash common.Hash
|
||||
if len(receipts) == 0 {
|
||||
|
|
@ -222,7 +208,7 @@ func (t *txAndReceipts) validateBlockReceipts(block *types.Block, receipts types
|
|||
if block.ExcessBlobGas() != nil {
|
||||
blobGasPrice = eip4844.CalcBlobFee(*block.ExcessBlobGas())
|
||||
}
|
||||
if err := newReceipts.DeriveFields(t.elConfig, block.Hash(), block.NumberU64(), block.Time(), block.BaseFee(), blobGasPrice, block.Transactions()); err != nil {
|
||||
if err := newReceipts.DeriveFields(c.elConfig, block.Hash(), block.NumberU64(), block.Time(), block.BaseFee(), blobGasPrice, block.Transactions()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// compare the JSON encoding of received and derived versions
|
||||
|
|
@ -240,9 +226,9 @@ func (t *txAndReceipts) validateBlockReceipts(block *types.Block, receipts types
|
|||
return newReceipts, nil
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) requestTxByHash(ctx context.Context, hash common.Hash) (tx *types.Transaction, isPending bool, err error) {
|
||||
func (c *Client) requestTxByHash(ctx context.Context, hash common.Hash) (tx *types.Transaction, isPending bool, err error) {
|
||||
var json *rpcTransaction
|
||||
err = t.client.CallContext(ctx, &json, "eth_getTransactionByHash", hash)
|
||||
err = c.client.CallContext(ctx, &json, "eth_getTransactionByHash", hash)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
} else if json == nil {
|
||||
|
|
@ -258,33 +244,33 @@ func (t *txAndReceipts) requestTxByHash(ctx context.Context, hash common.Hash) (
|
|||
return json.tx, json.BlockNumber == nil, nil
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) 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
|
||||
err := t.client.CallContext(ctx, &r, "eth_getTransactionReceipt", txHash)
|
||||
err := c.client.CallContext(ctx, &r, "eth_getTransactionReceipt", txHash)
|
||||
if err == nil && r == nil {
|
||||
return nil, ethereum.NotFound
|
||||
}
|
||||
return r, err
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) 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
|
||||
err := t.client.CallContext(ctx, &r, "eth_getBlockReceipts", blockHash)
|
||||
err := c.client.CallContext(ctx, &r, "eth_getBlockReceipts", blockHash)
|
||||
if err == nil && r == nil {
|
||||
return nil, ethereum.NotFound
|
||||
}
|
||||
return types.Receipts(r), err
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) cacheBlockTxPositions(block *types.Block) {
|
||||
func (c *Client) cacheBlockTxPositions(block *types.Block) {
|
||||
blockNumber, blockHash := block.NumberU64(), block.Hash()
|
||||
for i, tx := range block.Transactions() {
|
||||
t.txPosCache.Add(tx.Hash(), txInBlock{blockNumber: blockNumber, blockHash: blockHash, index: uint(i)})
|
||||
c.txPosCache.Add(tx.Hash(), txInBlock{blockNumber: blockNumber, blockHash: blockHash, index: uint(i)})
|
||||
}
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) sendTransaction(ctx context.Context, tx *types.Transaction) error {
|
||||
sender, err := types.Sender(t.signer, tx)
|
||||
func (c *Client) sendTransaction(ctx context.Context, tx *types.Transaction) error {
|
||||
sender, err := types.Sender(c.signer, tx)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -292,69 +278,69 @@ func (t *txAndReceipts) sendTransaction(ctx context.Context, tx *types.Transacti
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := t.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
|
||||
}
|
||||
t.sentTxLock.Lock()
|
||||
t.markTxAsSeen(sender, tx)
|
||||
t.sentTxLock.Unlock()
|
||||
c.sentTxLock.Lock()
|
||||
c.markTxAsSeen(sender, tx)
|
||||
c.sentTxLock.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) markTxAsSeen(sender common.Address, tx *types.Transaction) {
|
||||
senderTxs := t.sentTxs[sender]
|
||||
func (c *Client) markTxAsSeen(sender common.Address, tx *types.Transaction) {
|
||||
senderTxs := c.sentTxs[sender]
|
||||
if senderTxs == nil {
|
||||
senderTxs = make(map[common.Hash]sentTx)
|
||||
t.sentTxs[sender] = senderTxs
|
||||
c.sentTxs[sender] = senderTxs
|
||||
}
|
||||
senderTxs[tx.Hash()] = sentTx{nonce: tx.Nonce(), lastSeen: t.headCounter}
|
||||
senderTxs[tx.Hash()] = sentTx{nonce: tx.Nonce(), lastSeen: c.headCounter}
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) newHead(number uint64, hash common.Hash) {
|
||||
t.sentTxLock.Lock()
|
||||
if number > t.lastHeadNumber {
|
||||
t.lastHeadNumber = number
|
||||
t.headCounter++
|
||||
t.discardOldTxs()
|
||||
func (c *Client) txAndReceiptsNewHead(number uint64, hash common.Hash) {
|
||||
c.sentTxLock.Lock()
|
||||
if number > c.lastHeadNumber {
|
||||
c.lastHeadNumber = number
|
||||
c.headCounter++
|
||||
c.discardOldTxs()
|
||||
}
|
||||
t.sentTxLock.Unlock()
|
||||
c.sentTxLock.Unlock()
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) discardOldTxs() {
|
||||
for sender, senderTxs := range t.sentTxs {
|
||||
func (c *Client) discardOldTxs() {
|
||||
for sender, senderTxs := range c.sentTxs {
|
||||
for txHash, sentTx := range senderTxs {
|
||||
if sentTx.lastSeen+maxTxAge < t.headCounter {
|
||||
if sentTx.lastSeen+maxTxAge < c.headCounter {
|
||||
delete(senderTxs, txHash)
|
||||
}
|
||||
}
|
||||
if len(senderTxs) == 0 {
|
||||
delete(t.sentTxs, sender)
|
||||
delete(c.sentTxs, sender)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) allSenders() []common.Address {
|
||||
t.sentTxLock.Lock()
|
||||
allSenders := make([]common.Address, 0, len(t.sentTxs))
|
||||
for sender := range t.sentTxs {
|
||||
func (c *Client) allSenders() []common.Address {
|
||||
c.sentTxLock.Lock()
|
||||
allSenders := make([]common.Address, 0, len(c.sentTxs))
|
||||
for sender := range c.sentTxs {
|
||||
allSenders = append(allSenders, sender)
|
||||
}
|
||||
t.sentTxLock.Unlock()
|
||||
c.sentTxLock.Unlock()
|
||||
return allSenders
|
||||
}
|
||||
|
||||
func (t *txAndReceipts) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionHeader, sender common.Address) (uint64, types.Transactions, error) {
|
||||
proof, err := t.lightState.fetchProof(ctx, proofRequest{blockNumber: head.BlockNumber(), address: sender, storageKeys: ""})
|
||||
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 {
|
||||
return 0, nil, err
|
||||
}
|
||||
if err := t.lightState.validateProof(proof, head.StateRoot(), sender, nil); err != nil {
|
||||
if err := c.validateProof(proof, head.StateRoot(), sender, nil); err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
|
||||
t.sentTxLock.Lock()
|
||||
senderTxs := t.sentTxs[sender]
|
||||
t.sentTxLock.Unlock()
|
||||
c.sentTxLock.Lock()
|
||||
senderTxs := c.sentTxs[sender]
|
||||
c.sentTxLock.Unlock()
|
||||
|
||||
type pendingResult struct {
|
||||
pendingTx *types.Transaction
|
||||
|
|
@ -369,7 +355,7 @@ func (t *txAndReceipts) nonceAndPendingTxs(ctx context.Context, head *btypes.Exe
|
|||
}
|
||||
reqCount++
|
||||
go func() {
|
||||
tx, isPending, err := t.getUncachedTxByHash(ctx, txHash, head.BlockNumber())
|
||||
tx, isPending, err := c.getUncachedTxByHash(ctx, txHash, head.BlockNumber())
|
||||
if err == nil && isPending {
|
||||
resultCh <- pendingResult{pendingTx: tx}
|
||||
} else {
|
||||
|
|
@ -388,10 +374,10 @@ func (t *txAndReceipts) nonceAndPendingTxs(ctx context.Context, head *btypes.Exe
|
|||
}
|
||||
}
|
||||
sort.Sort(types.TxByNonce(pendingList))
|
||||
t.sentTxLock.Lock()
|
||||
c.sentTxLock.Lock()
|
||||
for _, tx := range pendingList {
|
||||
t.markTxAsSeen(sender, tx)
|
||||
c.markTxAsSeen(sender, tx)
|
||||
}
|
||||
t.sentTxLock.Unlock()
|
||||
c.sentTxLock.Unlock()
|
||||
return proof.Nonce, pendingList, nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue