core, eth, internal: split txIndexProgress from the main loop

core/blockchain_reader.go

@@ -17,7 +17,6 @@
 package core
 
 import (
-	"context"
 	"errors"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -299,8 +298,8 @@ func (bc *BlockChain) GetTransactionLookup(hash common.Hash) (*rawdb.LegacyTxLoo
 }
 
 // TxIndexDone returns true if the transaction indexer has finished indexing.
-func (bc *BlockChain) TxIndexDone(ctx context.Context) bool {
+func (bc *BlockChain) TxIndexDone() bool {
-	progress, err := bc.TxIndexProgress(ctx)
+	progress, err := bc.TxIndexProgress()
 	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
@@ -403,11 +402,11 @@ func (bc *BlockChain) GetVMConfig() *vm.Config {
 }
 
 // TxIndexProgress returns the transaction indexing progress.
-func (bc *BlockChain) TxIndexProgress(ctx context.Context) (TxIndexProgress, error) {
+func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) {
 	if bc.txIndexer == nil {
 		return TxIndexProgress{}, errors.New("tx indexer is not enabled")
 	}
-	return bc.txIndexer.txIndexProgress(ctx)
+	return bc.txIndexer.txIndexProgress(), nil
 }
 
 // HistoryPruningCutoff returns the configured history pruning point.

core/txindexer.go

@@ -17,9 +17,8 @@
 package core
 
 import (
-	"context"
-	"errors"
 	"fmt"
+	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
@@ -48,26 +47,35 @@ type txIndexer struct {
 	//       and all others shouldn't.
 	limit uint64
 
+	// The current tail of the indexed transactions, null indicates
+	// that no transactions have been indexed yet.
+	//
+	// This field is accessed by both the indexer and the indexing
+	// progress queries.
+	tail atomic.Pointer[uint64]
+
 	// cutoff denotes the block number before which the chain segment should
 	// be pruned and not available locally.
-	cutoff   uint64
+	cutoff uint64
-	db       ethdb.Database
+	chain  *BlockChain
-	progress chan chan TxIndexProgress
+	db     ethdb.Database
-	term     chan chan struct{}
+	term   chan chan struct{}
-	closed   chan struct{}
+	closed chan struct{}
 }
 
 // newTxIndexer initializes the transaction indexer.
 func newTxIndexer(limit uint64, chain *BlockChain) *txIndexer {
 	cutoff, _ := chain.HistoryPruningCutoff()
 	indexer := &txIndexer{
-		limit:    limit,
+		limit:  limit,
-		cutoff:   cutoff,
+		cutoff: cutoff,
-		db:       chain.db,
+		chain:  chain,
-		progress: make(chan chan TxIndexProgress),
+		db:     chain.db,
-		term:     make(chan chan struct{}),
+		term:   make(chan chan struct{}),
-		closed:   make(chan struct{}),
+		closed: make(chan struct{}),
 	}
+	indexer.tail.Store(rawdb.ReadTxIndexTail(chain.db))
+
 	go indexer.loop(chain)
 
 	var msg string
@@ -217,10 +225,9 @@ func (indexer *txIndexer) loop(chain *BlockChain) {
 
 	// Listening to chain events and manipulate the transaction indexes.
 	var (
-		stop     chan struct{}                       // Non-nil if background routine is active
+		stop chan struct{}           // Non-nil if background routine is active
-		done     chan struct{}                       // Non-nil if background routine is active
+		done chan struct{}           // Non-nil if background routine is active
-		head     = indexer.resolveHead()             // The latest announced chain head
+		head = indexer.resolveHead() // The latest announced chain head
-		lastTail = rawdb.ReadTxIndexTail(indexer.db) // The oldest indexed block, nil means nothing indexed
 
 		headCh = make(chan ChainHeadEvent)
 		sub    = chain.SubscribeChainHeadEvent(headCh)
@@ -245,13 +252,10 @@ func (indexer *txIndexer) loop(chain *BlockChain) {
 				done = make(chan struct{})
 				go indexer.run(h.Header.Number.Uint64(), stop, done)
 			}
-			head = h.Header.Number.Uint64()
 		case <-done:
 			stop = nil
 			done = nil
-			lastTail = rawdb.ReadTxIndexTail(indexer.db)
+			indexer.tail.Store(rawdb.ReadTxIndexTail(indexer.db))
-		case ch := <-indexer.progress:
-			ch <- indexer.report(head, lastTail)
 		case ch := <-indexer.term:
 			if stop != nil {
 				close(stop)
@@ -302,25 +306,12 @@ func (indexer *txIndexer) report(head uint64, tail *uint64) TxIndexProgress {
 	}
 }
 
-// txIndexProgress retrieves the tx indexing progress, or an error if the
+// txIndexProgress retrieves the transaction indexing progress. The reported
-// background tx indexer is already stopped.
+// progress may slightly lag behind the actual indexing state, as the tail is
-func (indexer *txIndexer) txIndexProgress(ctx context.Context) (TxIndexProgress, error) {
+// only updated at the end of each indexing operation. However, this delay is
-	ch := make(chan TxIndexProgress, 1)
+// considered acceptable.
-	select {
+func (indexer *txIndexer) txIndexProgress() TxIndexProgress {
-	case indexer.progress <- ch:
+	return indexer.report(indexer.resolveHead(), indexer.tail.Load())
-		select {
-		case prog := <-ch:
-			return prog, nil
-		case <-ctx.Done():
-			// Since the channel is buffered the loop will not block
-			// eventually when it prepares the response.
-			return TxIndexProgress{}, ctx.Err()
-		}
-	case <-indexer.closed:
-		return TxIndexProgress{}, errors.New("indexer is closed")
-	case <-ctx.Done():
-		return TxIndexProgress{}, ctx.Err()
-	}
 }
 
 // close shutdown the indexer. Safe to be called for multiple times.

core/txindexer_test.go

@@ -121,9 +121,8 @@ func TestTxIndexer(t *testing.T) {
 
 		// Index the initial blocks from ancient store
 		indexer := &txIndexer{
-			limit:    0,
+			limit: 0,
-			db:       db,
+			db:    db,
-			progress: make(chan chan TxIndexProgress),
 		}
 		for i, limit := range c.limits {
 			indexer.limit = limit
@@ -241,9 +240,8 @@ func TestTxIndexerRepair(t *testing.T) {
 
 		// Index the initial blocks from ancient store
 		indexer := &txIndexer{
-			limit:    c.limit,
+			limit: c.limit,
-			db:       db,
+			db:    db,
-			progress: make(chan chan TxIndexProgress),
 		}
 		indexer.run(chainHead, make(chan struct{}), make(chan struct{}))
 
@@ -432,13 +430,9 @@ func TestTxIndexerReport(t *testing.T) {
 
 		// Index the initial blocks from ancient store
 		indexer := &txIndexer{
-			limit:    c.limit,
+			limit:  c.limit,
-			cutoff:   c.cutoff,
+			cutoff: c.cutoff,
-			db:       db,
+			db:     db,
-			progress: make(chan chan TxIndexProgress),
-		}
-		if c.tail != nil {
-			rawdb.WriteTxIndexTail(db, *c.tail)
 		}
 		p := indexer.report(c.head, c.tail)
 		if p.Indexed != c.expIndexed {

eth/api_backend.go

@@ -361,8 +361,8 @@ func (b *EthAPIBackend) GetTransaction(txHash common.Hash) (bool, *types.Transac
 }
 
 // TxIndexDone returns true if the transaction indexer has finished indexing.
-func (b *EthAPIBackend) TxIndexDone(ctx context.Context) bool {
+func (b *EthAPIBackend) TxIndexDone() bool {
-	return b.eth.blockchain.TxIndexDone(ctx)
+	return b.eth.blockchain.TxIndexDone()
 }
 
 func (b *EthAPIBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {
@@ -391,7 +391,7 @@ func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.S
 
 func (b *EthAPIBackend) SyncProgress(ctx context.Context) ethereum.SyncProgress {
 	prog := b.eth.Downloader().Progress()
-	if txProg, err := b.eth.blockchain.TxIndexProgress(ctx); err == nil {
+	if txProg, err := b.eth.blockchain.TxIndexProgress(); err == nil {
 		prog.TxIndexFinishedBlocks = txProg.Indexed
 		prog.TxIndexRemainingBlocks = txProg.Remaining
 	}

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(context.Background()); err == nil {
+			if txProg, err := api.chain.TxIndexProgress(); err == nil {
 				prog.TxIndexFinishedBlocks = txProg.Indexed
 				prog.TxIndexRemainingBlocks = txProg.Remaining
 			}

eth/tracers/api.go

@@ -83,7 +83,7 @@ type Backend interface {
 	BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
 	BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error)
 	GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64)
-	TxIndexDone(ctx context.Context) bool
+	TxIndexDone() bool
 	RPCGasCap() uint64
 	ChainConfig() *params.ChainConfig
 	Engine() consensus.Engine
@@ -862,7 +862,7 @@ func (api *API) TraceTransaction(ctx context.Context, hash common.Hash, config *
 	found, _, blockHash, blockNumber, index := api.backend.GetTransaction(hash)
 	if !found {
 		// Warn in case tx indexer is not done.
-		if !api.backend.TxIndexDone(ctx) {
+		if !api.backend.TxIndexDone() {
 			return nil, ethapi.NewTxIndexingError()
 		}
 		// Only mined txes are supported

eth/tracers/api_test.go

@@ -121,7 +121,7 @@ func (b *testBackend) GetTransaction(txHash common.Hash) (bool, *types.Transacti
 	return tx != nil, tx, hash, blockNumber, index
 }
 
-func (b *testBackend) TxIndexDone(ctx context.Context) bool {
+func (b *testBackend) TxIndexDone() bool {
 	return true
 }
 

ethclient/ethclient_test.go

@@ -119,7 +119,7 @@ func newTestBackend(config *node.Config) (*node.Node, []*types.Block, error) {
 	}
 	// Ensure the tx indexing is fully generated
 	for ; ; time.Sleep(time.Millisecond * 100) {
-		progress, err := ethservice.BlockChain().TxIndexProgress(context.Background())
+		progress, err := ethservice.BlockChain().TxIndexProgress()
 		if err == nil && progress.Done() {
 			break
 		}

internal/ethapi/api.go

@@ -1340,7 +1340,7 @@ func (api *TransactionAPI) GetTransactionByHash(ctx context.Context, hash common
 			return NewRPCPendingTransaction(tx, api.b.CurrentHeader(), api.b.ChainConfig()), nil
 		}
 		// If also not in the pool there is a chance the tx indexer is still in progress.
-		if !api.b.TxIndexDone(ctx) {
+		if !api.b.TxIndexDone() {
 			return nil, NewTxIndexingError()
 		}
 		// If the transaction is not found in the pool and the indexer is done, return nil
@@ -1362,7 +1362,7 @@ func (api *TransactionAPI) GetRawTransactionByHash(ctx context.Context, hash com
 			return tx.MarshalBinary()
 		}
 		// If also not in the pool there is a chance the tx indexer is still in progress.
-		if !api.b.TxIndexDone(ctx) {
+		if !api.b.TxIndexDone() {
 			return nil, NewTxIndexingError()
 		}
 		// If the transaction is not found in the pool and the indexer is done, return nil
@@ -1376,7 +1376,7 @@ func (api *TransactionAPI) GetTransactionReceipt(ctx context.Context, hash commo
 	found, tx, blockHash, blockNumber, index := api.b.GetTransaction(hash)
 	if !found {
 		// Make sure indexer is done.
-		if !api.b.TxIndexDone(ctx) {
+		if !api.b.TxIndexDone() {
 			return nil, NewTxIndexingError()
 		}
 		// No such tx.
@@ -1786,7 +1786,7 @@ func (api *DebugAPI) GetRawTransaction(ctx context.Context, hash common.Hash) (h
 			return tx.MarshalBinary()
 		}
 		// If also not in the pool there is a chance the tx indexer is still in progress.
-		if !api.b.TxIndexDone(ctx) {
+		if !api.b.TxIndexDone() {
 			return nil, NewTxIndexingError()
 		}
 		// Transaction is not found in the pool and the indexer is done.

internal/ethapi/api_test.go

@@ -595,7 +595,7 @@ func (b testBackend) GetTransaction(txHash common.Hash) (bool, *types.Transactio
 	tx, blockHash, blockNumber, index := rawdb.ReadTransaction(b.db, txHash)
 	return true, tx, blockHash, blockNumber, index
 }
-func (b testBackend) TxIndexDone(ctx context.Context) bool {
+func (b testBackend) TxIndexDone() bool {
 	return true
 }
 func (b testBackend) GetPoolTransactions() (types.Transactions, error)         { panic("implement me") }

internal/ethapi/backend.go

@@ -75,7 +75,7 @@ type Backend interface {
 	// Transaction pool API
 	SendTx(ctx context.Context, signedTx *types.Transaction) error
 	GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64)
-	TxIndexDone(ctx context.Context) bool
+	TxIndexDone() bool
 	GetPoolTransactions() (types.Transactions, error)
 	GetPoolTransaction(txHash common.Hash) *types.Transaction
 	GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error)

internal/ethapi/transaction_args_test.go

@@ -383,7 +383,7 @@ func (b *backendMock) SendTx(ctx context.Context, signedTx *types.Transaction) e
 func (b *backendMock) GetTransaction(txHash common.Hash) (bool, *types.Transaction, common.Hash, uint64, uint64) {
 	return false, nil, [32]byte{}, 0, 0
 }
-func (b *backendMock) TxIndexDone(ctx context.Context) bool                     { return true }
+func (b *backendMock) TxIndexDone() bool                                        { return true }
 func (b *backendMock) GetPoolTransactions() (types.Transactions, error)         { return nil, nil }
 func (b *backendMock) GetPoolTransaction(txHash common.Hash) *types.Transaction { return nil }
 func (b *backendMock) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {