core, eth: minor fixes

core/blockchain.go

@@ -244,7 +244,6 @@ type BlockChain struct {
 	chainHeadFeed event.Feed
 	logsFeed      event.Feed
 	blockProcFeed event.Feed
-	txIndexFeed   event.Feed
 	scope         event.SubscriptionScope
 	genesisBlock  *types.Block
 
@@ -270,6 +269,8 @@ type BlockChain struct {
 	quit          chan struct{} // shutdown signal, closed in Stop.
 	stopping      atomic.Bool   // false if chain is running, true when stopped
 	procInterrupt atomic.Bool   // interrupt signaler for block processing
+
+	txIndexRunning bool                      // flag if the background tx indexer is activated
 	txIndexProgCh  chan chan TxIndexProgress // chan for querying the progress of transaction indexing
 
 	engine     consensus.Engine
@@ -487,6 +488,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 	// Start tx indexer/unindexer if required.
 	if txLookupLimit != nil {
 		bc.txLookupLimit = *txLookupLimit
+		bc.txIndexRunning = true
 
 		bc.wg.Add(1)
 		go bc.maintainTxIndex()
@@ -2462,7 +2464,7 @@ func (bc *BlockChain) reportTxIndexProgress(head uint64) TxIndexProgress {
 		indexed = head - *tail + 1
 	}
 	// The value of indexed might be larger than total if some blocks need
-	// to unindexed, avoiding a negative remaining.
+	// to be unindexed, avoiding a negative remaining.
 	if indexed < total {
 		remaining = total - indexed
 	}
@@ -2472,8 +2474,12 @@ func (bc *BlockChain) reportTxIndexProgress(head uint64) TxIndexProgress {
 	}
 }
 
-// TxIndexProgress retrieves the tx indexing progress.
+// TxIndexProgress retrieves the tx indexing progress, or an error if the
+// background tx indexer is not activated or already stopped.
 func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) {
+	if !bc.txIndexRunning {
+		return TxIndexProgress{}, errors.New("tx indexer is not activated")
+	}
 	ch := make(chan TxIndexProgress, 1)
 	select {
 	case bc.txIndexProgCh <- ch:
@@ -2527,14 +2533,6 @@ func (bc *BlockChain) maintainTxIndex() {
 			lastHead = head.Block.NumberU64()
 		case <-done:
 			done = nil
-
-			// WARNING, the event will be fired for each signal once the
-			// transaction indexing is finished. Subscribers need to manage
-			// the event stream by themselves. It's recommended to unsubscribe
-			// once the event is received.
-			if bc.reportTxIndexProgress(lastHead).Done() {
-				bc.txIndexFeed.Send(true)
-			}
 		case ch := <-bc.txIndexProgCh:
 			ch <- bc.reportTxIndexProgress(lastHead)
 		case <-bc.quit:

core/blockchain_reader.go

@@ -444,9 +444,3 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
 func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
 	return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
 }
-
-// SubscribeTxIndexEvent registers a subscription of bool where true means
-// transaction indexing has finished.
-func (bc *BlockChain) SubscribeTxIndexEvent(ch chan<- bool) event.Subscription {
-	return bc.scope.Track(bc.txIndexFeed.Subscribe(ch))
-}

eth/downloader/api.go

@@ -19,6 +19,7 @@ package downloader
 import (
 	"context"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/core"
@@ -57,25 +58,33 @@ func NewDownloaderAPI(d *Downloader, chain *core.BlockChain, m *event.TypeMux) *
 // new sync subscriptions and broadcasts sync status updates to the installed sync
 // subscriptions.
 //
-// The sync status pushed to subscriptions can a stream like:
+// The sync status pushed to subscriptions can be a stream like:
 // >>> {Syncing: true, Progress: {...}}
 // >>> {false}
 //
-// If the node is already synced up, then only a single event will be pushed {false}.
+// If the node is already synced up, then only a single event subscribers will
+// receive is {false}.
 func (api *DownloaderAPI) eventLoop() {
 	var (
 		sub               = api.mux.Subscribe(StartEvent{})
 		syncSubscriptions = make(map[chan interface{}]struct{})
-		txIndexCh         = make(chan bool, 1)
+		checkInterval     = time.Second * 30
+		checkTimer        = time.NewTimer(checkInterval)
+
+		// status flags
+		started bool
 		done    bool
-	)
-	txIndexSub := api.chain.SubscribeTxIndexEvent(txIndexCh)
-	defer func() {
-		if txIndexSub == nil {
-			return
+
+		getProgress = func() ethereum.SyncProgress {
+			prog := api.d.Progress()
+			if txProg, err := api.chain.TxIndexProgress(); err == nil {
+				prog.TxIndexFinishedBlocks = txProg.Indexed
+				prog.TxIndexRemainingBlocks = txProg.Remaining
 			}
-		txIndexSub.Unsubscribe()
-	}()
+			return prog
+		}
+	)
+	defer checkTimer.Stop()
 
 	for {
 		select {
@@ -93,11 +102,7 @@ func (api *DownloaderAPI) eventLoop() {
 			}
 			switch event.Data.(type) {
 			case StartEvent:
-				prog := api.d.Progress()
-				if txProg, err := api.chain.TxIndexProgress(); err == nil {
-					prog.TxIndexFinishedBlocks = txProg.Indexed
-					prog.TxIndexRemainingBlocks = txProg.Remaining
-				}
+				prog := getProgress()
 				notification := &SyncingResult{
 					Syncing: true,
 					Status:  prog,
@@ -106,30 +111,22 @@ func (api *DownloaderAPI) eventLoop() {
 				for c := range syncSubscriptions {
 					c <- notification
 				}
+				started = true
 			}
-		case synced := <-txIndexCh:
-			if !synced {
+		case <-checkTimer.C:
+			if !started {
+				checkTimer.Reset(checkInterval)
 				continue
 			}
-			prog := api.d.Progress()
+			prog := getProgress()
 			if !prog.Done() {
-				continue
-			}
-			txProg, err := api.chain.TxIndexProgress()
-			if err != nil || !txProg.Done() {
+				checkTimer.Reset(checkInterval)
 				continue
 			}
 			for c := range syncSubscriptions {
 				c <- false
 			}
 			done = true
-
-			// Unsubscribe the tx indexing events as the whole
-			// state sync is already finished.
-			if txIndexSub != nil {
-				txIndexSub.Unsubscribe()
-				txIndexSub = nil
-			}
 		}
 	}
 }