propogate context to txindexer

This commit is contained in:
Sina Mahmoodi 2025-05-01 15:47:38 +02:00
parent 9796c1bda1
commit ace9d07a1f
10 changed files with 27 additions and 19 deletions

View file

@ -17,6 +17,7 @@
package core package core
import ( import (
"context"
"errors" "errors"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
@ -277,7 +278,7 @@ func (bc *BlockChain) GetAncestor(hash common.Hash, number, ancestor uint64, max
// A null will be returned in the transaction is not found and background // A null will be returned in the transaction is not found and background
// transaction indexing is already finished. The transaction is not existent // transaction indexing is already finished. The transaction is not existent
// from the node's perspective. // from the node's perspective.
func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLookupEntry, *types.Transaction, error) { func (bc *BlockChain) GetTransactionLookup(ctx context.Context, hash common.Hash) (*rawdb.LegacyTxLookupEntry, *types.Transaction, error) {
bc.txLookupLock.RLock() bc.txLookupLock.RLock()
defer bc.txLookupLock.RUnlock() defer bc.txLookupLock.RUnlock()
@ -287,7 +288,7 @@ func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLoo
} }
tx, blockHash, blockNumber, txIndex := rawdb.ReadTransaction(bc.db, hash) tx, blockHash, blockNumber, txIndex := rawdb.ReadTransaction(bc.db, hash)
if tx == nil { if tx == nil {
progress, err := bc.TxIndexProgress() progress, err := bc.TxIndexProgress(ctx)
if err != nil { if err != nil {
// No error is returned if the transaction indexing progress is unreachable // No error is returned if the transaction indexing progress is unreachable
// due to unexpected internal errors. In such cases, it is impossible to // due to unexpected internal errors. In such cases, it is impossible to
@ -408,11 +409,11 @@ func (bc *BlockChain) GetVMConfig() *vm.Config {
} }
// TxIndexProgress returns the transaction indexing progress. // TxIndexProgress returns the transaction indexing progress.
func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) { func (bc *BlockChain) TxIndexProgress(ctx context.Context) (TxIndexProgress, error) {
if bc.txIndexer == nil { if bc.txIndexer == nil {
return TxIndexProgress{}, errors.New("tx indexer is not enabled") return TxIndexProgress{}, errors.New("tx indexer is not enabled")
} }
return bc.txIndexer.txIndexProgress() return bc.txIndexer.txIndexProgress(ctx)
} }
// HistoryPruningCutoff returns the configured history pruning point. // HistoryPruningCutoff returns the configured history pruning point.

View file

@ -17,6 +17,7 @@
package core package core
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
@ -303,13 +304,15 @@ func (indexer *txIndexer) report(head uint64, tail *uint64) TxIndexProgress {
// txIndexProgress retrieves the tx indexing progress, or an error if the // txIndexProgress retrieves the tx indexing progress, or an error if the
// background tx indexer is already stopped. // background tx indexer is already stopped.
func (indexer *txIndexer) txIndexProgress() (TxIndexProgress, error) { func (indexer *txIndexer) txIndexProgress(ctx context.Context) (TxIndexProgress, error) {
ch := make(chan TxIndexProgress, 1) ch := make(chan TxIndexProgress, 1)
select { select {
case indexer.progress <- ch: case indexer.progress <- ch:
return <-ch, nil return <-ch, nil
case <-indexer.closed: case <-indexer.closed:
return TxIndexProgress{}, errors.New("indexer is closed") return TxIndexProgress{}, errors.New("indexer is closed")
case <-ctx.Done():
return TxIndexProgress{}, ctx.Err()
} }
} }

View file

@ -357,7 +357,7 @@ func (b *EthAPIBackend) GetPoolTransaction(hash common.Hash) *types.Transaction
// indexing is already finished. The transaction is not existent from the perspective // indexing is already finished. The transaction is not existent from the perspective
// of node. // of node.
func (b *EthAPIBackend) GetTransaction(ctx context.Context, txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64, error) { func (b *EthAPIBackend) GetTransaction(ctx context.Context, txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64, error) {
lookup, tx, err := b.eth.blockchain.GetTransactionLookup(txHash) lookup, tx, err := b.eth.blockchain.GetTransactionLookup(ctx, txHash)
if err != nil { if err != nil {
return false, nil, common.Hash{}, 0, 0, err return false, nil, common.Hash{}, 0, 0, err
} }
@ -391,9 +391,9 @@ func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.S
return b.eth.txPool.SubscribeTransactions(ch, true) return b.eth.txPool.SubscribeTransactions(ch, true)
} }
func (b *EthAPIBackend) SyncProgress() ethereum.SyncProgress { func (b *EthAPIBackend) SyncProgress(ctx context.Context) ethereum.SyncProgress {
prog := b.eth.Downloader().Progress() prog := b.eth.Downloader().Progress()
if txProg, err := b.eth.blockchain.TxIndexProgress(); err == nil { if txProg, err := b.eth.blockchain.TxIndexProgress(ctx); err == nil {
prog.TxIndexFinishedBlocks = txProg.Indexed prog.TxIndexFinishedBlocks = txProg.Indexed
prog.TxIndexRemainingBlocks = txProg.Remaining prog.TxIndexRemainingBlocks = txProg.Remaining
} }

View file

@ -77,7 +77,7 @@ func (api *DownloaderAPI) eventLoop() {
getProgress = func() ethereum.SyncProgress { getProgress = func() ethereum.SyncProgress {
prog := api.d.Progress() prog := api.d.Progress()
if txProg, err := api.chain.TxIndexProgress(); err == nil { if txProg, err := api.chain.TxIndexProgress(context.Background()); err == nil {
prog.TxIndexFinishedBlocks = txProg.Indexed prog.TxIndexFinishedBlocks = txProg.Indexed
prog.TxIndexRemainingBlocks = txProg.Remaining prog.TxIndexRemainingBlocks = txProg.Remaining
} }

View file

@ -66,7 +66,7 @@ type backend interface {
CurrentHeader() *types.Header CurrentHeader() *types.Header
HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error) HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error)
Stats() (pending int, queued int) Stats() (pending int, queued int)
SyncProgress() ethereum.SyncProgress SyncProgress(ctx context.Context) ethereum.SyncProgress
} }
// fullNodeBackend encompasses the functionality necessary for a full node // fullNodeBackend encompasses the functionality necessary for a full node
@ -766,7 +766,7 @@ func (s *Service) reportStats(conn *connWrapper) error {
) )
// check if backend is a full node // check if backend is a full node
if fullBackend, ok := s.backend.(fullNodeBackend); ok { if fullBackend, ok := s.backend.(fullNodeBackend); ok {
sync := fullBackend.SyncProgress() sync := fullBackend.SyncProgress(context.Background())
syncing = !sync.Done() syncing = !sync.Done()
price, _ := fullBackend.SuggestGasTipCap(context.Background()) price, _ := fullBackend.SuggestGasTipCap(context.Background())
@ -775,7 +775,7 @@ func (s *Service) reportStats(conn *connWrapper) error {
gasprice += int(basefee.Uint64()) gasprice += int(basefee.Uint64())
} }
} else { } else {
sync := s.backend.SyncProgress() sync := s.backend.SyncProgress(context.Background())
syncing = !sync.Done() syncing = !sync.Done()
} }
// Assemble the node stats and send it to the server // Assemble the node stats and send it to the server

View file

@ -1530,8 +1530,8 @@ func (s *SyncState) TxIndexRemainingBlocks() hexutil.Uint64 {
// - healingBytecode: number of bytecodes pending // - healingBytecode: number of bytecodes pending
// - txIndexFinishedBlocks: number of blocks whose transactions are indexed // - txIndexFinishedBlocks: number of blocks whose transactions are indexed
// - txIndexRemainingBlocks: number of blocks whose transactions are not indexed yet // - txIndexRemainingBlocks: number of blocks whose transactions are not indexed yet
func (r *Resolver) Syncing() (*SyncState, error) { func (r *Resolver) Syncing(ctx context.Context) (*SyncState, error) {
progress := r.backend.SyncProgress() progress := r.backend.SyncProgress(ctx)
// Return not syncing if the synchronisation already completed // Return not syncing if the synchronisation already completed
if progress.Done() { if progress.Done() {

View file

@ -144,8 +144,8 @@ func (api *EthereumAPI) BlobBaseFee(ctx context.Context) *hexutil.Big {
// - highestBlock: block number of the highest block header this node has received from peers // - highestBlock: block number of the highest block header this node has received from peers
// - pulledStates: number of state entries processed until now // - pulledStates: number of state entries processed until now
// - knownStates: number of known state entries that still need to be pulled // - knownStates: number of known state entries that still need to be pulled
func (api *EthereumAPI) Syncing() (interface{}, error) { func (api *EthereumAPI) Syncing(ctx context.Context) (interface{}, error) {
progress := api.b.SyncProgress() progress := api.b.SyncProgress(ctx)
// Return not syncing if the synchronisation already completed // Return not syncing if the synchronisation already completed
if progress.Done() { if progress.Done() {

View file

@ -472,7 +472,9 @@ func (b *testBackend) setPendingBlock(block *types.Block) {
b.pending = block b.pending = block
} }
func (b testBackend) SyncProgress() ethereum.SyncProgress { return ethereum.SyncProgress{} } func (b testBackend) SyncProgress(ctx context.Context) ethereum.SyncProgress {
return ethereum.SyncProgress{}
}
func (b testBackend) SuggestGasTipCap(ctx context.Context) (*big.Int, error) { func (b testBackend) SuggestGasTipCap(ctx context.Context) (*big.Int, error) {
return big.NewInt(0), nil return big.NewInt(0), nil
} }

View file

@ -41,7 +41,7 @@ import (
// both full and light clients) with access to necessary functions. // both full and light clients) with access to necessary functions.
type Backend interface { type Backend interface {
// General Ethereum API // General Ethereum API
SyncProgress() ethereum.SyncProgress SyncProgress(ctx context.Context) ethereum.SyncProgress
SuggestGasTipCap(ctx context.Context) (*big.Int, error) SuggestGasTipCap(ctx context.Context) (*big.Int, error)
FeeHistory(ctx context.Context, blockCount uint64, lastBlock rpc.BlockNumber, rewardPercentiles []float64) (*big.Int, [][]*big.Int, []*big.Int, []float64, []*big.Int, []float64, error) FeeHistory(ctx context.Context, blockCount uint64, lastBlock rpc.BlockNumber, rewardPercentiles []float64) (*big.Int, [][]*big.Int, []*big.Int, []float64, []*big.Int, []float64, error)

View file

@ -323,7 +323,9 @@ func (b *backendMock) CurrentHeader() *types.Header { return b.current }
func (b *backendMock) ChainConfig() *params.ChainConfig { return b.config } func (b *backendMock) ChainConfig() *params.ChainConfig { return b.config }
// Other methods needed to implement Backend interface. // Other methods needed to implement Backend interface.
func (b *backendMock) SyncProgress() ethereum.SyncProgress { return ethereum.SyncProgress{} } func (b *backendMock) SyncProgress(ctx context.Context) ethereum.SyncProgress {
return ethereum.SyncProgress{}
}
func (b *backendMock) FeeHistory(ctx context.Context, blockCount uint64, lastBlock rpc.BlockNumber, rewardPercentiles []float64) (*big.Int, [][]*big.Int, []*big.Int, []float64, []*big.Int, []float64, error) { func (b *backendMock) FeeHistory(ctx context.Context, blockCount uint64, lastBlock rpc.BlockNumber, rewardPercentiles []float64) (*big.Int, [][]*big.Int, []*big.Int, []float64, []*big.Int, []float64, error) {
return nil, nil, nil, nil, nil, nil, nil return nil, nil, nil, nil, nil, nil, nil
} }