From ace9d07a1fe4f495ab7af24e924d2b5458baf91a Mon Sep 17 00:00:00 2001 From: Sina Mahmoodi Date: Thu, 1 May 2025 15:47:38 +0200 Subject: [PATCH] propogate context to txindexer --- core/blockchain_reader.go | 9 +++++---- core/txindexer.go | 5 ++++- eth/api_backend.go | 6 +++--- eth/downloader/api.go | 2 +- ethstats/ethstats.go | 6 +++--- graphql/graphql.go | 4 ++-- internal/ethapi/api.go | 4 ++-- internal/ethapi/api_test.go | 4 +++- internal/ethapi/backend.go | 2 +- internal/ethapi/transaction_args_test.go | 4 +++- 10 files changed, 27 insertions(+), 19 deletions(-) diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index b4ba5d9fd8..16404f4bf9 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -17,6 +17,7 @@ package core import ( + "context" "errors" "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 // transaction indexing is already finished. The transaction is not existent // 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() 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) if tx == nil { - progress, err := bc.TxIndexProgress() + progress, err := bc.TxIndexProgress(ctx) if err != nil { // No error is returned if the transaction indexing progress is unreachable // 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. -func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) { +func (bc *BlockChain) TxIndexProgress(ctx context.Context) (TxIndexProgress, error) { if bc.txIndexer == nil { 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. diff --git a/core/txindexer.go b/core/txindexer.go index 4f136a2d6f..7325095da6 100644 --- a/core/txindexer.go +++ b/core/txindexer.go @@ -17,6 +17,7 @@ package core import ( + "context" "errors" "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 // 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) select { case indexer.progress <- ch: return <-ch, nil case <-indexer.closed: return TxIndexProgress{}, errors.New("indexer is closed") + case <-ctx.Done(): + return TxIndexProgress{}, ctx.Err() } } diff --git a/eth/api_backend.go b/eth/api_backend.go index 10f7ffcbce..46cd2fb9dd 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -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 // of node. 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 { 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) } -func (b *EthAPIBackend) SyncProgress() ethereum.SyncProgress { +func (b *EthAPIBackend) SyncProgress(ctx context.Context) ethereum.SyncProgress { 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.TxIndexRemainingBlocks = txProg.Remaining } diff --git a/eth/downloader/api.go b/eth/downloader/api.go index ac175672a0..850f1a10a9 100644 --- a/eth/downloader/api.go +++ b/eth/downloader/api.go @@ -77,7 +77,7 @@ func (api *DownloaderAPI) eventLoop() { getProgress = func() ethereum.SyncProgress { 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.TxIndexRemainingBlocks = txProg.Remaining } diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index 0090a7d4c1..b6191baa12 100644 --- a/ethstats/ethstats.go +++ b/ethstats/ethstats.go @@ -66,7 +66,7 @@ type backend interface { CurrentHeader() *types.Header HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error) Stats() (pending int, queued int) - SyncProgress() ethereum.SyncProgress + SyncProgress(ctx context.Context) ethereum.SyncProgress } // 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 if fullBackend, ok := s.backend.(fullNodeBackend); ok { - sync := fullBackend.SyncProgress() + sync := fullBackend.SyncProgress(context.Background()) syncing = !sync.Done() price, _ := fullBackend.SuggestGasTipCap(context.Background()) @@ -775,7 +775,7 @@ func (s *Service) reportStats(conn *connWrapper) error { gasprice += int(basefee.Uint64()) } } else { - sync := s.backend.SyncProgress() + sync := s.backend.SyncProgress(context.Background()) syncing = !sync.Done() } // Assemble the node stats and send it to the server diff --git a/graphql/graphql.go b/graphql/graphql.go index 7af1adbb4a..0cb1d0cd54 100644 --- a/graphql/graphql.go +++ b/graphql/graphql.go @@ -1530,8 +1530,8 @@ func (s *SyncState) TxIndexRemainingBlocks() hexutil.Uint64 { // - healingBytecode: number of bytecodes pending // - txIndexFinishedBlocks: number of blocks whose transactions are indexed // - txIndexRemainingBlocks: number of blocks whose transactions are not indexed yet -func (r *Resolver) Syncing() (*SyncState, error) { - progress := r.backend.SyncProgress() +func (r *Resolver) Syncing(ctx context.Context) (*SyncState, error) { + progress := r.backend.SyncProgress(ctx) // Return not syncing if the synchronisation already completed if progress.Done() { diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index 3b699748b8..817d3f5a21 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -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 // - pulledStates: number of state entries processed until now // - knownStates: number of known state entries that still need to be pulled -func (api *EthereumAPI) Syncing() (interface{}, error) { - progress := api.b.SyncProgress() +func (api *EthereumAPI) Syncing(ctx context.Context) (interface{}, error) { + progress := api.b.SyncProgress(ctx) // Return not syncing if the synchronisation already completed if progress.Done() { diff --git a/internal/ethapi/api_test.go b/internal/ethapi/api_test.go index 5071e2412f..93dc11064a 100644 --- a/internal/ethapi/api_test.go +++ b/internal/ethapi/api_test.go @@ -472,7 +472,9 @@ func (b *testBackend) setPendingBlock(block *types.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) { return big.NewInt(0), nil } diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go index e28cb93296..449eaa2d59 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -41,7 +41,7 @@ import ( // both full and light clients) with access to necessary functions. type Backend interface { // General Ethereum API - SyncProgress() ethereum.SyncProgress + SyncProgress(ctx context.Context) ethereum.SyncProgress 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) diff --git a/internal/ethapi/transaction_args_test.go b/internal/ethapi/transaction_args_test.go index 9dd6a54729..b1c13fe73e 100644 --- a/internal/ethapi/transaction_args_test.go +++ b/internal/ethapi/transaction_args_test.go @@ -323,7 +323,9 @@ func (b *backendMock) CurrentHeader() *types.Header { return b.current } func (b *backendMock) ChainConfig() *params.ChainConfig { return b.config } // 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) { return nil, nil, nil, nil, nil, nil, nil }